/*
* This file is a part of the open-eBackup project.
* This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
* If a copy of the MPL was not distributed with this file, You can obtain one at
* http://mozilla.org/MPL/2.0/.
*
* Copyright (c) [2024] Huawei Technologies Co.,Ltd.
*
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
*/
#include "ObjectStorageRestoreJob.h"
#include "config_reader/ConfigIniReader.h"
#include "common/CleanMemPwd.h"
#include "common/EnvVarManager.h"
#include "client/ClientInvoke.h"
#include "constant/ErrorCode.h"
#include "client/ClientInvoke.h"
#include "ScanMgr.h"
#include "eSDKOBS.h"
#include "interface/CloudServiceInterface.h"
#include "interface/CloudServiceRequest.h"
#include "interface/CloudServiceResponse.h"
#include "manager/CloudServiceManager.h"
#include "utils/CertManager.h"

using namespace std;
using namespace Module;
namespace OBSPlugin {
namespace {
    constexpr auto MODULE = "ObjectStorageRestoreJob";
    constexpr uint32_t SCANNER_REPORT_CIRCLE_TIME = 60;  /* seconds */
    constexpr uint32_t NUMBER8 = 8;
    constexpr uint32_t NUMBER10 = 10;
    constexpr uint32_t NUMBER50 = 50;
    constexpr uint32_t NUMBER60 = 60;
    constexpr uint32_t NUMBER100 = 100;
    constexpr uint32_t NUMBER180 = 180;
    constexpr uint32_t NUMBER1024 = 1024;
    constexpr uint8_t INIT_INT_VALUE = 0;
    constexpr uint8_t RETRY_CNT = 3;
    constexpr uint32_t REPORT_RUNNING_INTERVAL = 10;
    constexpr uint32_t REPORT_RUNNING_TIMES = 6;
    constexpr uint32_t DEFAULT_AGGREGATE_NUM = 32;
    constexpr int MAX_AGGREGATE_NUM = 1000;
    const char SLASH_CH = '/';
    const std::string SCAN_SUB_JOB_RESTORE_INFO_FILE_NAME_SUFFIX = "_scan_restore_info.json";
}

#define ENTER                                                                                                   \
{                                                                                                               \
    m_mainJobRequestId = OBSPlugin::GenerateHash(m_jobId);                                                      \
    HCPTSP::getInstance().reset(m_mainJobRequestId);                                                            \
    HCP_Log(INFO, MODULE) << "Enter " << m_jobCtrlPhase                                                         \
                            << ", jobId: " << m_jobId                                                           \
                            << ", subjobId:" << ((m_subJobInfo == nullptr) ? "" : m_subJobId)                   \
                            << HCPENDLOG;                                                                       \
}

#define EXIT                                                                                                    \
{                                                                                                               \
    HCP_Log(INFO, MODULE) << "Exit " << m_jobCtrlPhase                                                          \
                            << ", jobId: " << m_jobId                                                           \
                            << ", subjobId:" << ((m_subJobInfo == nullptr) ? "" : m_subJobId)                   \
                            << HCPENDLOG;                                                                       \
}

bool ObjectStorageRestoreJob::GetRestoreJobInfo()
{
    if (GetJobInfo() != nullptr) {
        m_restoreJobPtr = dynamic_pointer_cast<AppProtect::RestoreJob>(GetJobInfo()->GetJobInfo());
    }
    if (m_restoreJobPtr == nullptr) {
        HCP_Log(ERR, MODULE) << "Failed to get restoreJobPtr." << HCPENDLOG;
        ReportJobDetailsWithLabelAndErrcode(
            std::make_tuple(JobLogLevel::TASK_LOG_ERROR, SubJobStatus::FAILED, OBSPlugin::PROGRESS0),
            "object_storage_plugin_restore_data_fail_label",
            ObsErrorCode::ERROR_AGENT_INTERNAL_ERROR);
        return false;
    }
    GetRestoreType();
    SetMainJobId(m_restoreJobPtr->jobId);
    SetSubJobId();
    InitJobInfo();
    return true;
}

ObjectStorageRestoreJob::~ObjectStorageRestoreJob()
{
    Module::CleanMemoryPwd(m_cacheFs.auth.authPwd);
    Module::CleanMemoryPwd(m_cacheFs.auth.extendInfo);
}

int ObjectStorageRestoreJob::PrerequisiteJob()
{
    SetJobCtrlPhase(OBSPlugin::JOB_CTRL_PHASE_PREJOB);
    if (!GetRestoreJobInfo()) {
        HCP_Log(ERR, MODULE) << "GetRestoreJobInfo failed" << HCPENDLOG;
        SetJobToFinish();
        return Module::FAILED;
    }
    if (!IsPigzExist()) {
        ReportJobDetailsWithLabelAndErrcode(
            std::make_tuple(JobLogLevel::TASK_LOG_ERROR, SubJobStatus::FAILED, OBSPlugin::PROGRESS0),
            "plugin_check_pigz_failed_label", 0);
        return Module::FAILED;
    }
    ENTER
    int ret = PrerequisiteJobInner();
    EXIT
    SetJobToFinish();
    return ret;
}

void ObjectStorageRestoreJob::HoldScanner()
{
    m_scanMgr.RegisterTask(ScanTaskLevel::REGULAR, m_subJobId);
    while (!m_scanMgr.HoldRunningTask(m_subJobId)) {
        HCP_Log(DEBUG, MODULE) << "HoldRunningTask" << HCPENDLOG;
        if (IsAbortJob()) {
            break;
        }
        Module::SleepFor(std::chrono::seconds(NUMBER10));
    }
}

void ObjectStorageRestoreJob::ReleaseScanner()
{
    if (!m_isScanParallel) {
        m_scanMgr.ReleaseRunningTask(m_subJobId);
    }
}

int ObjectStorageRestoreJob::GenerateSubJob()
{
    SetJobCtrlPhase(OBSPlugin::JOB_CTRL_PHASE_GENSUBJOB);
    if (!GetRestoreJobInfo()) {
        HCP_Log(ERR, MODULE) << "GetRestoreJobInfo failed" << HCPENDLOG;
        SetJobToFinish();
        return Module::FAILED;
    }
    int ret = Module::SUCCESS;
    ENTER
    m_generateSubjobFinish = false;
    // keep alive thread , used for report main job
    std::thread keepAlive = std::thread(&ObjectStorageRestoreJob::KeepPluginAlive, this);
    if (IsAbortJob()) {
        ERRLOG("Job aborted, skip scanner.");
    } else {
        // create sub job with esn
        if (!CreateSubJobTaskToCreateRestoreSubJob()) {
            HCP_Log(ERR, MODULE) << "CreateSubJobTaskToCreateRestoreSubJob failed" << HCPENDLOG;
            ret = Module::FAILED;
        }
    }
    m_generateSubjobFinish = true;
    if (keepAlive.joinable()) {
        keepAlive.join();
        INFOLOG("keep alive thread join!");
    }
    EXIT
    if (ret != Module::SUCCESS) {
        ReportJobDetailsWithLabel(SubJobStatus::FAILED, PROGRESS0, "", JobLogLevel::TASK_LOG_ERROR);
    } else {
        ReportJobDetailsWithLabel(SubJobStatus::COMPLETED, PROGRESS100, "", JobLogLevel::TASK_LOG_INFO);
    }
    SetJobToFinish();
    return ret;
}

int ObjectStorageRestoreJob::ExecuteSubJob()
{
    SetJobCtrlPhase(OBSPlugin::JOB_CTRL_PHASE_EXECSUBJOB);
    if (!GetRestoreJobInfo()) {
        HCP_Log(ERR, MODULE) << "GetRestoreJobInfo failed" << HCPENDLOG;
        SetJobToFinish();
        return Module::FAILED;
    }

    ENTER
    PrintSubJobInfo(m_subJobInfo);
    ObjectStorageBackupSubJob restoreSubJob {};
    if (!Module::JsonHelper::JsonStringToStruct(m_subJobInfo->jobInfo, restoreSubJob)) {
        HCP_Log(ERR, MODULE) << "Get restore subjob info failed" << HCPENDLOG;
        ReportJobDetailsWithLabelAndErrcode(
            std::make_tuple(JobLogLevel::TASK_LOG_ERROR, SubJobStatus::FAILED, OBSPlugin::PROGRESS0),
            "object_storage_plugin_restore_data_fail_label",
            ObsErrorCode::ERROR_AGENT_INTERNAL_ERROR);

        SetJobToFinish();
        return Module::FAILED;
    }
    m_subTaskType = restoreSubJob.m_SubTaskType;
    m_subJobRequestId = GenerateHash(m_jobId + m_subJobId);
    HCP_Log(INFO, MODULE) << "jobName: " << m_subJobInfo->jobName << ", jobId: " << m_jobId
                          << ", subJobId: " << m_subJobId << ", subJobRequestId: " << m_subJobRequestId << HCPENDLOG;
    // 当前待恢复子任务对应的副本ID以及ESN
    m_curCopyId = restoreSubJob.copyId;
    m_esn = restoreSubJob.esn;
    m_copyIndex = GetCopyRestoreOrder(restoreSubJob.copyId);

    bool keepAlive = true;
    std::thread keepSubJobAliveThread = std::thread(
        &ObjectStorageCommonService::KeepSubJobAlive, this, std::ref(keepAlive));
    int ret = Module::FAILED;
    if (m_subTaskType == OBSPlugin::SUBJOB_TYPE_PREFIX_SCAN_PHASE) {
        ret = ExecutePrefixScanSubJob(restoreSubJob);
    } else if (m_subTaskType == OBSPlugin::SUBJOB_TYPE_TEARDOWN_PHASE) {
        ret = ExecuteTeardownSubJobInner();
    } else if (m_subTaskType == OBSPlugin::SUBJOB_TYPE_SCAN_TEARDOWN_PHASE) {
        ret = ExecuteScanTeardownSubJob();
    } else if (m_subTaskType == SUBJOB_TYPE_CREATE_SUBJOB_PHASE) {
        ret = ExecuteCreateSubJob(restoreSubJob);
    } else if (m_subTaskType == OBSPlugin::SUBJOB_TYPE_ZERO_BACKUP_JUDGE_PHASE) {
        std::string filePath = m_cacheFsPath + "/obs-statistics/restore-stats-main-" + m_jobId + ".json";
        ret = ExecuteZeroBackupJudgeSubJob(filePath);
    } else {
        ret = ExecuteRestoreSubJobInner(restoreSubJob);
    }
    keepAlive = false;
    keepSubJobAliveThread.join();
    EXIT
    SetJobToFinish();
    return ret;
}

int ObjectStorageRestoreJob::PostJob()
{
    SetJobCtrlPhase(OBSPlugin::JOB_CTRL_PHASE_POSTJOB);
    if (!GetRestoreJobInfo()) {
        HCP_Log(ERR, MODULE) << "GetRestoreJobInfo failed" << HCPENDLOG;
        SetJobToFinish();
        return Module::FAILED;
    }
    ENTER
    int ret = PostJobInner();
    EXIT
    SetJobToFinish();
    return ret;
}

int ObjectStorageRestoreJob::PrerequisiteJobInner()
{
    int64_t errCode = ObsErrorCode::ERROR_AGENT_INTERNAL_ERROR;
    ABORT_ENDTASK(m_logSubJobDetails, m_logResult, m_logDetailList, m_logDetail, 0, 0);
    ReportJobDetailsWithLabel(SubJobStatus::RUNNING, OBSPlugin::PROGRESS0, "", JobLogLevel::TASK_LOG_INFO);
    if (m_dataLayoutExt.isNewCreateBucket == "true") {
        HCP_Log(INFO, MODULE) << "isNewCreateBucket" << HCPENDLOG;
        bool exit;
        if (CheckCreateBucketExist(exit) != Module::SUCCESS) {
            HCP_Log(ERR, MODULE) << "CheckCreateBucketExist failed" << HCPENDLOG;
            ReportJobDetailsWithLabelAndErrcode(
                std::make_tuple(JobLogLevel::TASK_LOG_ERROR, SubJobStatus::FAILED, OBSPlugin::PROGRESS0),
                "object_storage_plugin_restore_prepare_fail_label", errCode);
            return Module::FAILED;
        }
        if (exit) {
            ReportJobDetailsWithLabelAndErrcode(
                std::make_tuple(JobLogLevel::TASK_LOG_ERROR, SubJobStatus::FAILED, OBSPlugin::PROGRESS0),
                "object_storage_plugin_restore_bucket_exist_label", errCode);
            return Module::FAILED;
        }
    }
    PrintJobInfo();
    OBSPlugin::CreateDirectory(m_failureRecordRoot);
    if (!CreateSharedResources(m_jobId, m_generalInfo, m_scanStats, m_backupStats)) {
        HCP_Log(ERR, MODULE) << "CreateSharedResources failed" << HCPENDLOG;
        ReportJobDetailsWithLabelAndErrcode(
            std::make_tuple(JobLogLevel::TASK_LOG_ERROR, SubJobStatus::FAILED, OBSPlugin::PROGRESS0),
            "object_storage_plugin_restore_prepare_fail_label", errCode);
        return Module::FAILED;
    }
    m_generalInfo.m_jobStartTime = GetCurrentTimeInSeconds();
    if (!SetupMetaFsForRestoreJob()) {
        HCP_Log(ERR, MODULE) << "SetupMetaFsForRestoreJob failed" << HCPENDLOG;
        ReportJobDetailsWithLabelAndErrcode(
            std::make_tuple(JobLogLevel::TASK_LOG_ERROR, SubJobStatus::FAILED, OBSPlugin::PROGRESS0),
            "object_storage_plugin_restore_prepare_fail_label", errCode);
        return Module::FAILED;
    }
    if (!UpdateGeneralResource(m_jobId, m_generalInfo)) {
        HCP_Log(ERR, MODULE) << "UpdateGeneralResource failed" << HCPENDLOG;
        ReportJobDetailsWithLabelAndErrcode(
            std::make_tuple(JobLogLevel::TASK_LOG_ERROR, SubJobStatus::FAILED, OBSPlugin::PROGRESS0),
            "object_storage_plugin_restore_prepare_fail_label", errCode);
        return Module::FAILED;
    }
    ReportJobDetailsWithLabel(SubJobStatus::COMPLETED, OBSPlugin::PROGRESS100, "", JobLogLevel::TASK_LOG_INFO);
    return Module::SUCCESS;
}

void ObjectStorageRestoreJob::ReportJobCompleteStatus(
    SubJobStatus::type &jobStatus, std::string &jobLogLabel, int &jobProgress)
{
    int64_t errCode = ObsErrorCode::ERROR_AGENT_INTERNAL_ERROR;

    if (jobStatus == SubJobStatus::COMPLETED) {
        ReportJobDetails(JobLogLevel::TASK_LOG_INFO, SubJobStatus::COMPLETED, OBSPlugin::PROGRESS100);
    } else {
        if (m_scanStatus == SCANNER_STATUS::PROTECTED_SERVER_NOT_REACHABLE) {
            errCode = ObsErrorCode::ERROR_OBS_SCAN_PROTECTED_SERVER_NOT_REACHABLE;
        } else if (m_scanStatus == SCANNER_STATUS::SECONDARY_SERVER_NOT_REACHABLE) {
            errCode = ObsErrorCode::ERROR_OBS_SCAN_SECONDARY_SERVER_NOT_REACHABLE;
        }
        ReportJobDetailsWithLabelAndErrcode(
            std::make_tuple(JobLogLevel::TASK_LOG_ERROR, jobStatus, jobProgress), jobLogLabel, errCode);
    }
    return;
}

bool ObjectStorageRestoreJob::ReportStartScanInfo()
{
    std::string scanStatisticsPath = OBSPlugin::PathJoin(m_cacheFsPath, "scan-statistics");
    const std::string fileName = scanStatisticsPath + "/" + m_subJobId + SCAN_SUB_JOB_RESTORE_INFO_FILE_NAME_SUFFIX;
    if (IsFileExist(fileName)) {
        if (!JsonFileTool::ReadFromFile(fileName, m_scanStats)) {
            HCP_Log(INFO, MODULE) << "No scan restore info is read, file name " << fileName << HCPENDLOG;
        }
    }

    std::string scanStartTimePath = PathJoin(GetPathName(m_cacheFsPath), m_jobId, "restoreTime");
    HCP_Log(INFO, MODULE) << "ScanStartTimePath " << scanStartTimePath << HCPENDLOG;
    if (OBSPlugin::IsDirExist(scanStartTimePath)) {
        HCP_Log(INFO, MODULE) << "Not first path" << scanStartTimePath << HCPENDLOG;
    } else {
        CreateDirectory(scanStartTimePath);
        ReportJobDetailsWithLabel(SubJobStatus::RUNNING, OBSPlugin::PROGRESS0,
            "object_storage_plugin_restore_scan_start_label", JobLogLevel::TASK_LOG_INFO);
    }

    bool bReportToPM = false;
    if (!UpdateScanStatsResourceWithLock(m_scanStats, bReportToPM)) {
        HCP_Log(WARN, MODULE) << "Update Scan Stats failed after hold scanner" << HCPENDLOG;
        return false;
    }

    return true;
}


bool ObjectStorageRestoreJob::IsGrainRestoreOrPermanentIncrCopy()
{
    if (m_fineGrainedRestore ||
        (m_restoreJobPtr->copies[0].dataType == AppProtect::CopyDataType::PERMANENT_INCREMENTAL_COPY)) {
        return true;
    }
    return false;
}

int ObjectStorageRestoreJob::ProcessPrefixForScan(const PrefixInfo& prefixInfo, string metaPath, std::string esn)
{
    m_subfolder = GetScanGenFileRelativeDir(prefixInfo);
    if (!InitPathForRestore()) {
        HCP_Log(ERR, MODULE) << "InitPathForRestore failed" << HCPENDLOG;
        return Module::FAILED;
    }

    auto tempPrefix = prefixInfo;
    std::string prefixInfoStr;
    if (!Module::JsonHelper::StructToJsonString(tempPrefix, prefixInfoStr)) {
        HCP_Log(WARN, MODULE) << "Convert to json failed for prefix info:" << HCPENDLOG;
    }

    HCP_Log(INFO, MODULE) << "resotre prefix:" << prefixInfoStr << ", esn:" << esn << HCPENDLOG;
    if (!UnzipDcachefiles(metaPath, m_subfolder, m_dcaheAndFcachePath)) {
        HCP_Log(ERR, MODULE) << "unzip dcache and fcache failed" << HCPENDLOG;
        return Module::FAILED;
    }

    // 细粒度恢复,永久增量副本恢复也需要扫描解决ctrol和meta不匹配问题
    if (IsGrainRestoreOrPermanentIncrCopy()) {
        HCP_Log(INFO, MODULE) << "StartScanner dataType:" << m_restoreJobPtr->copies[0].dataType << HCPENDLOG;
        if (!StartScanner()) {
            return Module::FAILED;
        }
        MonitorScanner(prefixInfo, m_restoreJobStatus, m_restoreJobLogLabel, m_restoreJobProgress);
        HCP_Log(INFO, MODULE) << "Scan finished for prefix:" << prefixInfoStr << ", esn:" << esn << HCPENDLOG;
        ClearGenerateSubJobEnv();
    } else {
        // 全量恢复或者普通增量恢复
        std::string savedCtrlPathParent = OBSPlugin::PathJoin(metaPath, "ctrl", m_subfolder);
        std::vector<std::string> fileList{};
        if (!GetFileListInDirectory(savedCtrlPathParent, fileList)) {
            HCP_Log(ERR, MODULE) << "Get filelist for ctrl file failed: " << savedCtrlPathParent << HCPENDLOG;
            return Module::FAILED;
        }
        std::vector<std::string> paramList{};
        // 空桶场景，ctrl文件为空则跳过
        if (fileList.size() != 0) {
            std::string savedCtrlPath = OBSPlugin::PathJoin(metaPath, "ctrl", m_subfolder);
            std::string ctrlFileParentPath = OBSPlugin::GetPathName(m_scanControlFilePath);
            std::string copyCtrlCmd = "cp -r " + savedCtrlPath + " " + OBSPlugin::GetPathName(ctrlFileParentPath);
            HCP_Log(INFO, MODULE) << "Run copy control files cmd : " << copyCtrlCmd << HCPENDLOG;
            if (OBSPlugin::RunShellCmd(copyCtrlCmd, paramList) != Module::SUCCESS) {
                return Module::FAILED;
            }
        }

        std::string savedScanStatPath = OBSPlugin::PathJoin(metaPath, "scan-statistics");
        std::string copyScanStatisticsCmd = "cp -r " + savedScanStatPath + " " + m_cacheFsPath;
        HCP_Log(INFO, MODULE) << "Run copy scan-statistics cmd : " << copyScanStatisticsCmd << HCPENDLOG;
        if (OBSPlugin::RunShellCmd(copyScanStatisticsCmd, paramList) != Module::SUCCESS) {
            return Module::FAILED;
        }
    }

    return Module::SUCCESS;
}

int ObjectStorageRestoreJob::UnzipPreCacheFiles(const ObjectStorageBackupSubJob& restoreSubJob)
{
    auto preCopyId = m_restoreJobPtr->copies[m_copyIndex - 1].id;
    auto& copyMap = m_metaFsPathMap[preCopyId];
    for (auto iter : copyMap) {
        std::string esn = iter.first;
        HCP_Log(INFO, MODULE) << "copyId:" << preCopyId << ", esn: " << esn << HCPENDLOG;
        std::string preMetaPath = m_metaFsPathMap[preCopyId][esn];
        for (const auto& prefixInfo : restoreSubJob.GetFlatPrefixInfos()) {
            std::string subfolder = GetScanGenFileRelativeDir(prefixInfo);
            std::string preDcaheAndFcachePath = OBSPlugin::PathJoin(m_cacheFsPath, "restore-job", m_jobId,
                to_string(m_copyIndex - 1), "scan", "meta", subfolder);
            if (!OBSPlugin::RecurseCreateDirectory(preDcaheAndFcachePath)) {
                HCP_Log(ERR, MODULE) << "create prev dcache&fcache path failed" << HCPENDLOG;
                return Module::FAILED;
            }
            if (!UnzipDcachefiles(preMetaPath, subfolder, preDcaheAndFcachePath)) {
                HCP_Log(ERR, MODULE) << "unzip dcache and fcache failed" << HCPENDLOG;
                return Module::FAILED;
            }
        }
    }
    return Module::SUCCESS;
}

int ObjectStorageRestoreJob::CreateSubJobInner(const ObjectStorageBackupSubJob& restoreSubJob)
{
    ABORT_ENDTASK(m_logSubJobDetails, m_logResult, m_logDetailList, m_logDetail, 0, 0);
    if (!InitJobInfo()) {
        HCP_Log(ERR, MODULE) << "InitJobInfo failed" << HCPENDLOG;
        return Module::FAILED;
    }
    PrintJobInfo();

    if (!ReportStartScanInfo()) {
        return Module::FAILED;
    }

    m_copyIndex = GetCopyRestoreOrder(restoreSubJob.copyId);
    if (m_aggregateRestore && m_fineGrainedRestore && m_copyIndex > 0) {
        int ret = UnzipPreCacheFiles(restoreSubJob);
        if (ret != Module::SUCCESS) {
            return ret;
        }
    }
    std::string& metaPath = m_metaFsPathMap[restoreSubJob.copyId][restoreSubJob.esn];
    // 前缀信息预处理，一次只扫描一个前缀（subPrefixs平铺）
    for (const auto& prefixInfo : restoreSubJob.GetFlatPrefixInfos()) {
        int ret = ProcessPrefixForScan(prefixInfo, metaPath, restoreSubJob.esn);
        if (ret != Module::SUCCESS) {
            return ret;
        }
    }

    if (!m_fineGrainedRestore) {
        // 非细粒度恢复，不进行扫描，直接设置完成状态
        m_restoreJobStatus = SubJobStatus::COMPLETED;
    }
    if (!CreateFurtherSubTasks(restoreSubJob)) {
        ERRLOG("Create SubJob Failed!");
        return Module::FAILED;
    }
    ReportJobCompleteStatus(m_restoreJobStatus, m_restoreJobLogLabel, m_restoreJobProgress);
    return Module::SUCCESS;
}

void ObjectStorageRestoreJob::ClearGenerateSubJobEnv()
{
    if (m_scanner) {
        m_scanner->Destroy();
        m_scanner = nullptr;
    }
}

void ObjectStorageRestoreJob::GetRestoreType()
{
    HCP_Log(INFO, MODULE) << "Enter GetRestoreType:" << HCPENDLOG;
    if (m_restoreJobPtr->jobParam.restoreType == AppProtect::RestoreJobType::type::FINE_GRAINED_RESTORE) {
        m_fineGrainedRestore = true;
        HCP_Log(INFO, MODULE) << "=====This is FINE_GRAINED_RESTORE=====" << HCPENDLOG;
    }
}

void ObjectStorageRestoreJob::ReportJobProgress(SubJobStatus::type &jobStatus, std::string &jobLogLabel)
{
    if (jobStatus ==  SubJobStatus::COMPLETED) {
        ReportJobDetailsWithLabel(SubJobStatus::COMPLETED, OBSPlugin::PROGRESS100, "", JobLogLevel::TASK_LOG_INFO);
    } else {
        int64_t errCode = ObsErrorCode::ERROR_AGENT_INTERNAL_ERROR;
        switch (m_backupStatus) {
            case BackupPhaseStatus::FAILED_NOACCESS:
                errCode = ObsErrorCode::ERROR_BACKUP_FAILED_NOACCESS_ERROR;
                break;
            case BackupPhaseStatus::FAILED_NOSPACE:
                errCode = ObsErrorCode::ERROR_BACKUP_FAILED_NOSPACE_ERROR;
                break;
            case BackupPhaseStatus::FAILED_PROT_SERVER_NOTREACHABLE:
                errCode = ObsErrorCode::ERROR_OBS_BACKUP_PROTECTED_SERVER_NOT_REACHABLE;
                break;
            case BackupPhaseStatus::FAILED_SEC_SERVER_NOTREACHABLE:
                errCode = ObsErrorCode::ERROR_OBS_BACKUP_SECONDARY_SERVER_NOT_REACHABLE;
                break;
            default:
                errCode = ObsErrorCode::ERROR_AGENT_INTERNAL_ERROR;
        }
        if (!m_backErrMsg.empty() && errCode == ObsErrorCode::ERROR_AGENT_INTERNAL_ERROR) {
            ReportJobDetailsWithDetailAndErrcode(std::make_tuple(JobLogLevel::TASK_LOG_ERROR, jobStatus,
                OBSPlugin::PROGRESS0), jobLogLabel, errCode, m_backErrMsg);
        } else {
            ReportJobDetailsWithLabelAndErrcode(
                std::make_tuple(JobLogLevel::TASK_LOG_ERROR, jobStatus, OBSPlugin::PROGRESS0), jobLogLabel, errCode);
        }
    }

    return;
}

bool ObjectStorageRestoreJob::RetryUpdateBackupStartTime(ObjectStorageBackupSubJob& restoreSubJob)
{
    int retry = INIT_INT_VALUE;
    bool ret = false;
    while (!(ret = UpdateBackupStartTimeInSharedResource(restoreSubJob)) && retry < RETRY_CNT) {
        Module::SleepFor(std::chrono::seconds(NUMBER8 * ++retry));
    }
    return ret;
}

int ObjectStorageRestoreJob::ExecuteRestoreSubJobInner(ObjectStorageBackupSubJob restoreSubJob)
{
    ABORT_ENDTASK(m_logSubJobDetails, m_logResult, m_logDetailList, m_logDetail, 0, 0);
    ReportJobDetailsWithLabel(SubJobStatus::RUNNING, OBSPlugin::PROGRESS0, "", JobLogLevel::TASK_LOG_INFO);
    BackupStats backupStatistics {};
    int jobProgress = 0;
    int retryCnt = 0;
    SubJobStatus::type jobStatus = SubJobStatus::FAILED;
    std::string jobLogLabel = "object_storage_plugin_restore_data_fail_label";

    m_nodeLevelTaskInfo.Insert(m_jobId);
    m_nodeLevelTaskInfo.IncrSubTasksCount();
    if (!InitJobInfo()) {
        HCP_Log(ERR, MODULE) << "InitJobInfo failed" << HCPENDLOG;
        return HandleExecuteSubJobInner(backupStatistics, jobStatus, jobLogLabel);
    }
    PrintJobInfo();

    if (!GetSharedResources(m_jobId, m_generalInfo, m_scanStats, m_backupStats)) {
        HCP_Log(ERR, MODULE) << "GetSharedResources failed" << HCPENDLOG;
        return HandleExecuteSubJobInner(backupStatistics, jobStatus, jobLogLabel);
    }
    /* update the control file path */
    string controlFile = restoreSubJob.m_ControlFile;
    restoreSubJob.m_ControlFile = m_cacheFsPath + restoreSubJob.m_ControlFile;
    INFOLOG("control file is: %s", restoreSubJob.m_ControlFile.c_str());

    if (!RetryUpdateBackupStartTime(restoreSubJob)) {
        HCP_Log(ERR, MODULE) << "UpdateBackupStartTimeInSharedResource failed" << HCPENDLOG;
        return HandleExecuteSubJobInner(backupStatistics, jobStatus, jobLogLabel);
    }

    MONITOR_BACKUP_RES_TYPE monitorRet;
    do {
        if (!StartRestore(restoreSubJob)) {
            HCP_Log(ERR, MODULE) << "StartRestore failed" << HCPENDLOG;
            return HandleExecuteSubJobInner(backupStatistics, jobStatus, jobLogLabel);
        }
        monitorRet = MonitorRestore(backupStatistics, jobStatus, jobLogLabel, jobProgress);
        GetBackupFailDetails();
        DestroyBackup();
    } while (monitorRet == MONITOR_BACKUP_RES_TYPE::MONITOR_BACKUP_RES_TYPE_NEEDRETRY && ++retryCnt < RETRY_CNT);

    if (retryCnt >= RETRY_CNT && monitorRet == MONITOR_BACKUP_RES_TYPE::MONITOR_BACKUP_RES_TYPE_NEEDRETRY) {
        // seems this sub job is stuck for some reason , copy this control file to meta repo for further check
        INFOLOG("subjob is stuck, %s, copy controlFile: %s", m_subJobId.c_str(), restoreSubJob.m_ControlFile.c_str());
        OBSPlugin::CopyFile(restoreSubJob.m_ControlFile,
            m_metaFsPathMap[restoreSubJob.copyId][restoreSubJob.esn] + controlFile);
    }

    // for clean code, move to a funciton
    return HandleExecuteSubJobInner(backupStatistics, jobStatus, jobLogLabel);
}

int ObjectStorageRestoreJob::HandleExecuteSubJobInner(
    BackupStats &backupStatistics, SubJobStatus::type &jobStatus, std::string &jobLogLabel)
{
    m_nodeLevelTaskInfo.DecrSubTasksCount();
    m_dataSize = backupStatistics.noOfBytesCopied/NUMBER1024;
    ReportJobProgress(jobStatus, jobLogLabel);
    return Module::SUCCESS;
}

void ObjectStorageRestoreJob::DestroyBackup()
{
    if (m_backup != nullptr) {
        m_backup->Destroy();
        m_backup.reset();
    }
}

bool ObjectStorageRestoreJob::CreateFurtherSubTasks(const ObjectStorageBackupSubJob& subJob)
{
    if (m_idGenerator == nullptr) {
        InitIdGenerator();
    }
    std::string jobName = "ObsRestore_CreateSubJob" + std::to_string(m_idGenerator->GenerateId());
    ObjectStorageBackupSubJob subJobInfo = subJob;
    subJobInfo.m_SubTaskType = SUBJOB_TYPE_CREATE_SUBJOB_PHASE;
    std::string subJobStr;
    if (!Module::JsonHelper::StructToJsonString(subJobInfo, subJobStr)) {
        ERRLOG("Json Transfer Failed!");
        return false;
    }
    SubJob subTask {};
    subTask.__set_jobId(m_jobId);
    subTask.__set_jobType(SubJobType::BUSINESS_SUB_JOB);
    subTask.__set_policy(ExecutePolicy::ANY_NODE);
    subTask.__set_jobInfo(subJobStr);
    subTask.__set_ignoreFailed(false);
    subTask.__set_jobPriority(SUBJOB_TYPE_CREATE_SUB_JOB_PRIO);
    subTask.__set_jobName(jobName);

    do {
        if (IsAbortJob()) {
            WARNLOG("Exit received Abort for taskid: %s, subtaskid: %s", m_jobId.c_str(), m_subJobId.c_str());
            break;
        }
        int ret = CreateSubTask(subTask);
        if (ret == Module::SUCCESS) {
            break;
        } else if (ret == Module::RETRY) {
            HCP_Log(WARN, MODULE) << "Create subtask failed with retriable error"<< HCPENDLOG;
            SleepForCreateSubTaskError();
            continue;
        } else {
            ERRLOG("Exit CreateFurtherSubTasks, CreateSubtask failed");
            return false;
        }
    } while (true);
    return true;
}

int ObjectStorageRestoreJob::ExecutePrefixScanSubJob(const ObjectStorageBackupSubJob& restoreSubJob)
{
    HCP_Log(INFO, MODULE)  << "Enter ExecutePrefixScanSubJob"  << HCPENDLOG;
    int ret;

    int scanParallel = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION, "DME_OBS_SCAN_PARALLEL_SWITCH");
    m_isScanParallel = (scanParallel == 0) ? false : true;
    HCP_Log(INFO, MODULE) << "The scan parallel switch is " << m_isScanParallel << HCPENDLOG;

    std::shared_ptr<void> defer(nullptr, [&](...) {
        ReleaseScanner();
    });

    if (!m_isScanParallel) {
        HoldScanner();
    }
    m_scanStats.m_scanStartTime = OBSPlugin::GetCurrentTimeInSeconds();
    ret = CreateSubJobInner(restoreSubJob);
    if (ret != Module::SUCCESS) {
        HCP_Log(ERR, MODULE) << "ExecutePrefixScanSubJob failed jobId: " << m_jobId << HCPENDLOG;
        ReportJobDetailsWithLabelAndErrcode(
            std::make_tuple(JobLogLevel::TASK_LOG_ERROR, SubJobStatus::FAILED, OBSPlugin::PROGRESS0),
            "object_storage_plugin_restore_data_fail_label",
            ObsErrorCode::ERROR_AGENT_INTERNAL_ERROR);
    }

    return ret;
}

int ObjectStorageRestoreJob::ExecuteCreateSubJob(const ObjectStorageBackupSubJob &subJob)
{
    SubJobStatus::type jobStatus = SubJobStatus::COMPLETED;
    INFOLOG("Enter ExecuteCreateSubJob: %s, %s", m_jobId.c_str(), m_subJobId.c_str());
    for (const auto &prefixInfo : subJob.GetFlatPrefixInfos()) {
        if (!HandleMonitorScannerCompletion(prefixInfo, jobStatus)) {
            HCP_Log(ERR, MODULE) << "Handle monitor scanner completion failed" << HCPENDLOG;
            return Module::FAILED;
        }
    }
    INFOLOG("Exit ExecuteCreateSubJob: %s, %s", m_jobId.c_str(), m_subJobId.c_str());
    return Module::SUCCESS;
}

int ObjectStorageRestoreJob::ExecuteTeardownSubJobInner()
{
    HCP_Log(INFO, MODULE)  << "Enter ExecuteTeardownSubJobInner"  << HCPENDLOG;
    ObjectStorageNativeBackupStats backupStatistics {};
    CalculateJobStats();
    std::string filePath = m_cacheFsPath + "/obs-statistics/restore-stats-main-" + m_jobId + ".json";
    ReadBackupStatsFromFile(filePath, backupStatistics);
    PrintBackupCopyInfo(backupStatistics);
    ReportBackupCompletionStatus(backupStatistics);

    return Module::SUCCESS;
}
int ObjectStorageRestoreJob::ExecuteScanTeardownSubJob()
{
    std::string scanStatisticsPath = OBSPlugin::PathJoin(m_cacheFsPath, "scan-statistics");
    ObjectStorageNativeScanStatistics mainJobStats {};
    bool ret = false;
    if (!IsGrainRestoreOrPermanentIncrCopy()) {
        ret = GetBackupScanStats(scanStatisticsPath, mainJobStats);
    } else {
        ret = CalcSubScanStats(scanStatisticsPath, true, mainJobStats);
    }
    if (!ret) {
        return Module::FAILED;
    }
    std::string filePath = scanStatisticsPath + "/scan-stats-main-" + m_jobId + ".json";
    if (!JsonFileTool::WriteToFile(mainJobStats, filePath)) {
        HCP_Log(ERR, MODULE) << "Write to file failed: " << filePath << HCPENDLOG;
        return Module::FAILED;
    }
    ReportScannerCompleteStatus(mainJobStats);
    return Module::SUCCESS;
}

void ObjectStorageRestoreJob::RemoveCacheDirectories()
{
    OBSPlugin::Remove(m_cacheFsPath);
    return;
}

bool ObjectStorageRestoreJob::CalculateJobStats()
{
    uint64_t noOfDirCopiedFromDirMtimePhase = 0;
    uint32_t noOfFiles = 0;
    ObjectStorageNativeBackupStats mainJobStats {};

    std::string dir = OBSPlugin::PathJoin(m_cacheFsPath, "obs-statistics");
    std::vector<std::string> fileList {};
    if (!GetFileListInDirectory(dir, fileList)) {
        HCP_Log(ERR, MODULE) << "Get filelist for dir failed: " << dir << HCPENDLOG;
        return false;
    }
    for (const std::string& path : fileList) {
        if (noOfFiles++ % NUMBER50 == 0) {
            /* Since this loop may take too much time, report progress to framework, every 50 files */
            ReportJobDetailsWithLabel(SubJobStatus::RUNNING, OBSPlugin::PROGRESS0, "", JobLogLevel::TASK_LOG_INFO);
        }
        ObjectStorageNativeBackupStats subJobStats {};
        ReadBackupStatsFromFile(path, subJobStats);
        mainJobStats += subJobStats;
        RemoveFile(path);
    }

    std::string filePath = m_cacheFsPath + "/obs-statistics/restore-stats-main-" + m_jobId + ".json";
    WriteBackupStatsToFile(filePath, mainJobStats);
    return true;
}

int ObjectStorageRestoreJob::PostJobInner()
{
    ReportJobDetailsWithLabel(SubJobStatus::RUNNING, OBSPlugin::PROGRESS0, "", JobLogLevel::TASK_LOG_INFO);
    if (!InitJobInfo()) {
        RemoveCacheDirectories();
        DeleteSharedResources(m_jobId);
        ReportJobDetailsWithLabelAndErrcode(
            std::make_tuple(JobLogLevel::TASK_LOG_ERROR, SubJobStatus::FAILED, OBSPlugin::PROGRESS0),
            "",
            ObsErrorCode::ERROR_AGENT_INTERNAL_ERROR);
        m_nodeLevelTaskInfo.Erase(m_jobId);
        return Module::FAILED;
    }
    PrintJobInfo();
    MergeBackupFailureRecords();
    if (!GetSharedResources(m_jobId, m_generalInfo, m_scanStats, m_backupStats)) {
        RemoveCacheDirectories();
        DeleteSharedResources(m_jobId);
        ReportJobDetailsWithLabelAndErrcode(
            std::make_tuple(JobLogLevel::TASK_LOG_ERROR, SubJobStatus::FAILED, OBSPlugin::PROGRESS0),
            "",
            ObsErrorCode::ERROR_AGENT_INTERNAL_ERROR);
        m_nodeLevelTaskInfo.Erase(m_jobId);
        return Module::FAILED;
    }
    CalculateJobStats();
    // Delete all the shared reources which are created in the PrerequisiteJob
    DeleteSharedResources(m_jobId);
    RemoveCacheDirectories();
    ReportJobDetailsWithLabel(SubJobStatus::COMPLETED, OBSPlugin::PROGRESS100, "", JobLogLevel::TASK_LOG_INFO);
    m_nodeLevelTaskInfo.Erase(m_jobId);
    return Module::SUCCESS;
}

bool ObjectStorageRestoreJob::GenerateSubJobList(const PrefixInfo& prefixInfo, std::vector<SubJob> &subJobList,
    std::vector<std::string> &ctrlFileList, const std::string &srcCtrlFileFullPath,
    const std::string &dstCtrlFileFullPath)
{
    SubJob subJob {};
    uint32_t subTaskType {};
    string dstCtrlFileRelPath = dstCtrlFileFullPath.substr(m_cacheFsPath.length(), string::npos);
    if (!InitSubJobInfo(prefixInfo, subJob, m_jobId, dstCtrlFileRelPath, subTaskType)) {
        HCP_Log(ERR, MODULE) << "Init subtask failed" << HCPENDLOG;
        return false;
    }
    subJobList.push_back(subJob);
    ctrlFileList.push_back(srcCtrlFileFullPath);

    return true;
}

bool ObjectStorageRestoreJob::CreateSubTasksFromCtrlFile(const PrefixInfo& prefixInfo, std::string srcDir,
    std::string dstDir, uint32_t subTaskType)
{
    std::vector<SubJob> subJobList {};
    std::vector<std::string> ctrlFileList {};
    uint32_t validCtrlFileCntr = 0;
    static int64_t lastCreateJobErrTime = 0;

    lastCreateJobErrTime = 0;
    std::vector<std::string> srcFileList {};
    if (!checkFilePathAndGetSrcFileList(srcDir, dstDir, srcFileList)) {
        return false;
    }
    INFOLOG("Sub task type: %u, NumOfCtrlFiles: %d", subTaskType, srcFileList.size());

    for (uint32_t i = 0; i < srcFileList.size(); ++i) {
        if (IsAbortJob()) {
            INFOLOG("Exit Abort for taskid: %s, subtaskid: %s", m_jobId.c_str(), m_subJobId.c_str());
            return true;
        }
        std::string srcCtrlFileFullPath = srcFileList[i];
        if (!IsValidCtrlFile(subTaskType, srcCtrlFileFullPath)) {
            continue;
        }

        std::string ctrlFileParentPath = GetPathName(srcCtrlFileFullPath);
        std::string ctrlFileRelPath = srcCtrlFileFullPath.substr(m_cacheFsPath.length(), string::npos);
        std::string ctrlFile = srcCtrlFileFullPath.substr(ctrlFileParentPath.length() + NUMBER1,
            srcCtrlFileFullPath.length() - ctrlFileParentPath.length() - NUMBER1);
        std::string dstCtrlFileFullPath = OBSPlugin::PathJoin(dstDir, ctrlFile);
        OBSPlugin::CopyFile(srcCtrlFileFullPath, dstCtrlFileFullPath);
        if (!GenerateSubJobList(prefixInfo, subJobList, ctrlFileList, srcCtrlFileFullPath, dstCtrlFileFullPath)) {
            HCP_Log(ERR, MODULE) << "Exit, Create sub tasks failed" << HCPENDLOG;
            return false;
        }
        // We create 10 Jobs at a time. If 10 is not accumulated, continue
        if (subJobList.size() % NUMBER10 != 0) {
            continue;
        }
        if (!CheckAndRetryCreateSubTask(subJobList, ctrlFileList, lastCreateJobErrTime, true, false)) {
            return false;
        }
    }
    if (!CheckAndRetryCreateSubTask(subJobList, ctrlFileList, lastCreateJobErrTime, true, false)) {
        return false;
    }

    HCP_Log(INFO, MODULE) << "Exit. Create sub tasks success." << HCPENDLOG;
    return true;
}

bool ObjectStorageRestoreJob::UpdateCopyPhaseStartTimeInGenRsc()
{
    if (m_generalInfo.m_backupCopyPhaseStartTime != 0) {
        return true;
    }

    HCP_Log(INFO, MODULE) << "Enter UpdateCopyPhaseStartTimeInGenRsc" << HCPENDLOG;
    m_generalInfo.m_backupCopyPhaseStartTime = GetCurrentTimeFromRemoteServer(m_cacheFsPath);
    if (m_generalInfo.m_backupCopyPhaseStartTime == 0) {
        HCP_Log(ERR, MODULE) << "Get current time of first backup copy subtask failed" << HCPENDLOG;
        return false;
    }
    if (!(UpdateGeneralResource(m_jobId, m_generalInfo))) {
        HCP_Log(ERR, MODULE) << "UpdateGeneralResource failed" << HCPENDLOG;
        return false;
    }
    ReportJobDetailsWithLabel(SubJobStatus::RUNNING, OBSPlugin::PROGRESS0,
        "object_storage_plugin_restore_data_start_label", JobLogLevel::TASK_LOG_INFO);
    HCP_Log(INFO, MODULE) << "Exit UpdateCopyPhaseStartTimeInGenRsc" << HCPENDLOG;
    return true;
}

bool ObjectStorageRestoreJob::InitSubJobInfo(const PrefixInfo& prefixInfo, SubJob &subJob, const std::string jobId,
    const std::string ctrlFile, uint32_t &subTaskType)
{
    std::string subTaskName;
    uint32_t subTaskPrio {0};
    ObjectStorageBackupSubJob restoreSubJobInfo;

    GetSubJobTypeByFileName(ctrlFile, subTaskName, subTaskType, subTaskPrio);
    restoreSubJobInfo.m_ControlFile = ctrlFile;
    restoreSubJobInfo.m_SubTaskType = subTaskType;
    restoreSubJobInfo.esn = m_esn;
    restoreSubJobInfo.copyId = m_curCopyId;
    restoreSubJobInfo.prefixInfo.emplace_back(prefixInfo);

    restoreSubJobInfo.dataCachePath = ""; // 子任务进行拼接
    restoreSubJobInfo.metaFilePath = "";
    restoreSubJobInfo.checkpointCachePath = m_checkPointPath;
    // dcache和fcache解压路径
    restoreSubJobInfo.dcacheAndFcachePath = OBSPlugin::PathJoin(m_cacheFsPath, "restore-job", m_jobId,
        to_string(m_copyIndex), "scan", "meta", GetScanGenFileRelativeDir(prefixInfo));
    restoreSubJobInfo.dcacheAndFcachePath =
        restoreSubJobInfo.dcacheAndFcachePath.substr(m_cacheFsPath.length(), string::npos);
    string restoreSubJobInfoStr;
    if (!Module::JsonHelper::StructToJsonString(restoreSubJobInfo, restoreSubJobInfoStr)) {
        HCP_Log(ERR, MODULE) << "Convert to json failed for subJob info: " << HCPENDLOG;
        return false;
    }
    HCP_Log(INFO, MODULE) << "restoreSubJobInfoStr is : "  << restoreSubJobInfoStr << HCPENDLOG;
    subTaskPrio = GetCopyRestoreOrder(m_curCopyId) * SUBJOB_TYPE_AGGREGATE_BASE_PRIO + subTaskPrio;
    subJob.__set_jobId(jobId);
    subJob.__set_jobName(subTaskName);
    subJob.__set_jobType(SubJobType::BUSINESS_SUB_JOB);
    subJob.__set_policy(ExecutePolicy::ANY_NODE);
    subJob.__set_jobInfo(restoreSubJobInfoStr);
    subJob.__set_jobPriority(subTaskPrio);
    subJob.__set_ignoreFailed(false);
    return true;
}

void ObjectStorageRestoreJob::MonitorScanner(const PrefixInfo& prefixInfo,
    SubJobStatus::type &jobStatus, std::string &jobLogLabel, int &jobProgress)
{
    HCP_Log(INFO, MODULE) << "Enter Monitor Scanner" << HCPENDLOG;
    SCANNER_TASK_STATUS scanTaskStatus = SCANNER_TASK_STATUS::SCANNER_TASK_STATUS_INPROGRESS;
    jobStatus = SubJobStatus::RUNNING;
    jobProgress = 0;
    jobLogLabel = "";
    ObjectStorageNativeScanStatistics scanStatistics;
    do {
        m_scanStatus = m_scanner->GetStatus();
        /* Ensure scanner is ready and start to scan */
        if (m_scanStatus == SCANNER_STATUS::INIT) {
            Module::SleepFor(std::chrono::seconds(OBSPlugin::SUBTASK_WAIT_FOR_SCANNER_READY_IN_SEC));
            continue;
        }
        UpdateScannerStatistics(m_scanStats, scanStatistics);
        FillMonitorScannerVarDetails(scanTaskStatus, jobStatus, jobLogLabel, jobProgress);
        UpdateScanStatInfo(prefixInfo, scanStatistics);
        if (scanTaskStatus != SCANNER_TASK_STATUS::SCANNER_TASK_STATUS_INPROGRESS) {
            break;
        }

        if (IsAbortJob()) {
            HCP_Log(INFO, MODULE) << "Scanner - Abort is invocked for"
                                  << " taskid: " << m_jobId << ", subtaskid: " << m_subJobId << HCPENDLOG;
            if (m_scanner->Abort() != SCANNER_STATUS::SUCCESS) {
                HCP_Log(ERR, MODULE) << "scanner Abort is failed" << HCPENDLOG;
            }
        }

        HCP_Log(INFO, MODULE) << "scanner status:" << (int)scanTaskStatus << HCPENDLOG;
        Module::SleepFor(std::chrono::seconds(OBSPlugin::GENERATE_SUBTASK_MONITOR_DUR_IN_SEC));
    } while (true);

    m_scanStats = scanStatistics;
    m_scanStats.m_scannedPrefixInfo.emplace_back(prefixInfo);
    std::string scanStatisticsPath = OBSPlugin::PathJoin(m_cacheFsPath, "scan-statistics");
    const std::string fileName = scanStatisticsPath + "/" + m_subJobId + SCAN_SUB_JOB_RESTORE_INFO_FILE_NAME_SUFFIX;
    if (!JsonFileTool::WriteToFile(m_scanStats, fileName)) {
        HCP_Log(WARN, MODULE) << "Write scanStatistics failed, file name " << fileName << HCPENDLOG;
    }
    HCP_Log(INFO, MODULE) << "Exit Monitor Scanner" << HCPENDLOG;
}

bool ObjectStorageRestoreJob::HandleMonitorScannerCompletion(
    const PrefixInfo& prefixInfo, SubJobStatus::type &jobStatus)
{
    if (jobStatus != SubJobStatus::COMPLETED) {
        return true;
    }
    const std::string scanCtrlFilePath = OBSPlugin::PathJoin(m_cacheFsPath, "restore-job", m_jobId,
        to_string(m_copyIndex), "scan", "control", GetScanGenFileRelativeDir(prefixInfo));
    const std::string bakCtrlFilePath = OBSPlugin::PathJoin(m_cacheFsPath, "restore-job", m_jobId,
        to_string(m_copyIndex), "restore", "control", GetScanGenFileRelativeDir(prefixInfo));
    if (!CreateSubTasksFromCtrlFile(prefixInfo, scanCtrlFilePath, bakCtrlFilePath, SUBJOB_TYPE_DATACOPY_COPY_PHASE)) {
        HCP_Log(ERR, MODULE) << "Create subtask failed" << HCPENDLOG;
        return false;
    }
    if (!CreateSubTasksFromCtrlFile(prefixInfo, scanCtrlFilePath, bakCtrlFilePath, SUBJOB_TYPE_DATACOPY_DELETE_PHASE)) {
        HCP_Log(ERR, MODULE) << "Create subtask failed" << HCPENDLOG;
        return false;
    }
    return true;
}

bool ObjectStorageRestoreJob::ReportScannerCompleteStatus(const ObjectStorageNativeScanStatistics &scanStatistics)
{
    if (scanStatistics.m_totFailedFiles != 0) {
        ReportJobDetailsWithLabel(SubJobStatus::FAILED,
            OBSPlugin::PROGRESS100,
            "object_storage_plugin_restore_scan_completed_with_warn_label",
            JobLogLevel::TASK_LOG_WARNING,
            std::to_string(scanStatistics.m_totFilesToBackup),
            OBSPlugin::FormatCapacity(scanStatistics.m_totalSizeToBackup),
            std::to_string(scanStatistics.m_totFailedFiles),
            std::to_string(scanStatistics.m_totFilesToBackup),
            FormatCapacity(scanStatistics.m_totalSizeToBackup));
    } else {
        ReportJobDetailsWithLabel(SubJobStatus::COMPLETED,
            OBSPlugin::PROGRESS100,
            "object_storage_plugin_restore_scan_completed_label",
            JobLogLevel::TASK_LOG_INFO,
            std::to_string(scanStatistics.m_totFilesToBackup),
            OBSPlugin::FormatCapacity(scanStatistics.m_totalSizeToBackup),
            std::to_string(scanStatistics.m_totFilesToBackup),
            FormatCapacity(scanStatistics.m_totalSizeToBackup));
    }
    return true;
}

void ObjectStorageRestoreJob::FillMonitorScannerVarDetails(SCANNER_TASK_STATUS &scanTaskStatus,
    SubJobStatus::type &jobStatus, std::string &jobLogLabel, int &jobProgress)
{
    if (m_scanStatus == SCANNER_STATUS::COMPLETED) {
        HCP_Log(INFO, MODULE) << "Scan completed" << HCPENDLOG;
        jobProgress = OBSPlugin::PROGRESS100;
        jobStatus = SubJobStatus::COMPLETED;
        jobLogLabel = "object_storage_plugin_restore_scan_completed_label";
        scanTaskStatus = SCANNER_TASK_STATUS::SCANNER_TASK_STATUS_SUCCESS;
    } else if (m_scanStatus == SCANNER_STATUS::FAILED) {
        HCP_Log(ERR, MODULE) << "Scan failed" << HCPENDLOG;
        jobStatus = SubJobStatus::FAILED;
        jobLogLabel = "object_storage_plugin_restore_scan_fail_label";
        scanTaskStatus = SCANNER_TASK_STATUS::SCANNER_TASK_STATUS_FAILED;
    } else if (m_scanStatus == SCANNER_STATUS::ABORT_IN_PROGRESS) {
        HCP_Log(ERR, MODULE) << "Scan abort in progress" << HCPENDLOG;
        jobStatus = SubJobStatus::ABORTING;
        jobLogLabel = "";
    } else if (m_scanStatus == SCANNER_STATUS::ABORTED) {
        HCP_Log(ERR, MODULE) << "Scan aborted" << HCPENDLOG;
        jobStatus = SubJobStatus::ABORTED;
        jobLogLabel = "";
        scanTaskStatus = SCANNER_TASK_STATUS::SCANNER_TASK_STATUS_ABORTED;
    } else if (m_scanStatus == SCANNER_STATUS::SCAN_READ_COMPLETED) {
        jobStatus = SubJobStatus::RUNNING;
        jobLogLabel = "";
    } else if (m_scanStatus == SCANNER_STATUS::CTRL_DIFF_IN_PROGRESS) {
        jobStatus = SubJobStatus::RUNNING;
        jobLogLabel = "";
    } else if (m_scanStatus == SCANNER_STATUS::SECONDARY_SERVER_NOT_REACHABLE) {
        HCP_Log(ERR, MODULE) << "Scan failed as sec nas server is not reachable" << HCPENDLOG;
        jobStatus = SubJobStatus::FAILED;
        jobLogLabel = "object_storage_plugin_restore_scan_fail_label";
        scanTaskStatus = SCANNER_TASK_STATUS::SCANNER_TASK_STATUS_FAILED;
    } else if (m_scanStatus == SCANNER_STATUS::PROTECTED_SERVER_NOT_REACHABLE) {
        HCP_Log(ERR, MODULE) << "Scan failed as protected nas server is not reachable" << HCPENDLOG;
        jobStatus = SubJobStatus::FAILED;
        jobLogLabel = "object_storage_plugin_restore_scan_fail_label";
        scanTaskStatus = SCANNER_TASK_STATUS::SCANNER_TASK_STATUS_FAILED;
    }
    return;
}

ObjectStorageRestoreJob::MONITOR_BACKUP_RES_TYPE ObjectStorageRestoreJob::MonitorRestore(BackupStats &backupStatistics,
    SubJobStatus::type &jobStatus, std::string &jobLogLabel, int &jobProgress)
{
    HCP_Log(INFO, MODULE) << "Enter Monitor Backup" << HCPENDLOG;
    jobStatus = SubJobStatus::RUNNING;
    jobProgress = 0;
    jobLogLabel = "";
    BackupStats tmpStats;
    time_t statLastUpdateTime = OBSPlugin::GetCurrentTimeInSeconds();
    int threadTime =  GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION, "BACKUP_STUCK_TIME");

    do {
        m_backupStatus = m_backup->GetStatus();
        HCP_Log(INFO, MODULE) << "backupStatus:" << static_cast<int>(m_backupStatus) << HCPENDLOG;
        tmpStats = m_backup->GetStats();
        if (backupStatistics != tmpStats) {
            statLastUpdateTime = OBSPlugin::GetCurrentTimeInSeconds();
            INFOLOG("backup statistics last update time: %ld", statLastUpdateTime);
            backupStatistics = tmpStats;
        }
        if (m_backupStatus == BackupPhaseStatus::INPROGRESS &&
            OBSPlugin::GetCurrentTimeInSeconds() - statLastUpdateTime > threadTime) {
            HandleMonitorStuck(backupStatistics, jobStatus, jobLogLabel, jobProgress);
            return MONITOR_BACKUP_RES_TYPE::MONITOR_BACKUP_RES_TYPE_NEEDRETRY;
        }
        UpdateBackupStatistics(backupStatistics);
        if (m_backupStatus == BackupPhaseStatus::COMPLETED) {
            HandleMonitorComplete(jobStatus, jobLogLabel, jobProgress);
            return MONITOR_BACKUP_RES_TYPE::MONITOR_BACKUP_RES_TYPE_SUCCESS;
        } else if (m_backupStatus == BackupPhaseStatus::FAILED ||
                   m_backupStatus == BackupPhaseStatus::FAILED_NOACCESS ||
                   m_backupStatus == BackupPhaseStatus::FAILED_NOSPACE ||
                   m_backupStatus == BackupPhaseStatus::FAILED_SEC_SERVER_NOTREACHABLE ||
                   m_backupStatus == BackupPhaseStatus::FAILED_PROT_SERVER_NOTREACHABLE) {
            HandleMonitorFailed(jobStatus, jobLogLabel);
            return MONITOR_BACKUP_RES_TYPE::MONITOR_BACKUP_RES_TYPE_FAILED;
        } else if (m_backupStatus == BackupPhaseStatus::ABORTED) {
            HCP_Log(ERR, MODULE) << "Monitor Backup - BACKUP_PHASE_STATUS_ABORTED " << HCPENDLOG;
            jobStatus = SubJobStatus::ABORTED;
            jobLogLabel = "";
            return MONITOR_BACKUP_RES_TYPE::MONITOR_BACKUP_RES_TYPE_ABORTED;
        } else if (m_backupStatus == BackupPhaseStatus::ABORT_INPROGRESS) {
            jobStatus = SubJobStatus::ABORTING;
            jobLogLabel = "";
        }

        if (IsAbortJob()) {
            HandleMonitorAbort();
        }
        Module::SleepFor(std::chrono::seconds(OBSPlugin::GENERATE_SUBTASK_MONITOR_DUR_IN_SEC));
    } while (true);
    HCP_Log(INFO, MODULE) << "Exit Monitor Backup" << HCPENDLOG;
    return MONITOR_BACKUP_RES_TYPE::MONITOR_BACKUP_RES_TYPE_SUCCESS;
}

void ObjectStorageRestoreJob::HandleMonitorAbort()
{
    HCP_Log(INFO, MODULE) << "Restore - Abort is invocked for"
        << " taskid: " << m_jobId << ", subtaskid: " << m_subJobId << HCPENDLOG;
    if (BackupRetCode::SUCCESS != m_backup->Abort()) {
        HCP_Log(ERR, MODULE) << "restore sub task  Abort is failed" << HCPENDLOG;
    }
}

void ObjectStorageRestoreJob::HandleMonitorComplete(
    SubJobStatus::type &jobStatus, std::string &jobLogLabel, int &jobProgress)
{
    HCP_Log(INFO, MODULE) << "Monitor Backup - BACKUP_PHASE_STATUS_COMPLETED " << HCPENDLOG;
    jobLogLabel = "";
    jobProgress = OBSPlugin::PROGRESS100;
    jobStatus = SubJobStatus::COMPLETED;
}

void ObjectStorageRestoreJob::HandleMonitorFailed(SubJobStatus::type &jobStatus, std::string &jobLogLabel)
{
    HCP_Log(ERR, MODULE) << "Monitor Backup - BACKUP_PHASE_STATUS_FAILED " << HCPENDLOG;
    jobStatus = SubJobStatus::FAILED;
    jobLogLabel = "object_storage_plugin_restore_data_fail_label";
}

void ObjectStorageRestoreJob::HandleMonitorStuck(BackupStats &backupStatistics,
    SubJobStatus::type &jobStatus, std::string &jobLogLabel, int &jobProgress)
{
    WARNLOG("backup statistic has not been update for 300s");
    if (BackupRetCode::SUCCESS != m_backup->Abort()) {
        HCP_Log(ERR, MODULE) << "backup Abort is failed" << HCPENDLOG;
    }
    backupStatistics.noOfDirFailed += backupStatistics.noOfDirToBackup - backupStatistics.noOfDirCopied;
    backupStatistics.noOfFilesFailed += backupStatistics.noOfFilesToBackup - backupStatistics.noOfFilesCopied;
    UpdateBackupStatistics(backupStatistics);
    jobLogLabel = "";
    jobProgress = OBSPlugin::PROGRESS100;
    jobStatus = SubJobStatus::COMPLETED;
}

void ObjectStorageRestoreJob::GetBackupFailDetails()
{
    std::unordered_set<FailedRecordItem, FailedRecordItemHash> failedRecordItem = m_backup->GetFailedDetails();
    if (!failedRecordItem.empty()) {
        auto it = failedRecordItem.begin();
        m_backErrMsg = it->errMsg;
        HCP_Log(ERR, MODULE) << "m_backErrMsg" << m_backErrMsg.c_str() << HCPENDLOG;
    }
    HCP_Log(DEBUG, MODULE) << "No Erorr Message" << HCPENDLOG;
}

bool ObjectStorageRestoreJob::ReportBackupRunningStatus()
{
    ObjectStorageNativeBackupStats mainBackupJobStatistics {};
    std::string dir = OBSPlugin::PathJoin(m_cacheFsPath, "obs-statistics");
    std::vector<std::string> fileList {};
    if (!GetFileListInDirectory(dir, fileList)) {
        HCP_Log(ERR, MODULE) << "Get filelist for dir failed: " << dir << HCPENDLOG;
        return false;
    }
    uint32_t i = 0;
    uint32_t noOfFiles = 0;
    for (i = 0; i < fileList.size(); ++i) {
        ObjectStorageNativeBackupStats subBackupJobStatistics {};
        if (!ReadBackupStatsFromFile(fileList[i], subBackupJobStatistics))
            continue;
        mainBackupJobStatistics.m_noOfFilesCopied += subBackupJobStatistics.m_noOfFilesCopied;
        mainBackupJobStatistics.m_noOfBytesCopied += subBackupJobStatistics.m_noOfBytesCopied;
    }

    if (mainBackupJobStatistics.m_noOfBytesCopied != 0) {
        ReportJobDetailsWithLabel(SubJobStatus::RUNNING,
            OBSPlugin::PROGRESS0,
            "object_storage_plugin_restore_data_inprogress_label",
            JobLogLevel::TASK_LOG_INFO,
            std::to_string(mainBackupJobStatistics.m_noOfFilesCopied),
            FormatCapacity(mainBackupJobStatistics.m_noOfBytesCopied));
    } else {
        ReportJobDetailsWithLabel(SubJobStatus::RUNNING, OBSPlugin::PROGRESS0, "", JobLogLevel::TASK_LOG_INFO);
    }
    return true;
}

bool ObjectStorageRestoreJob::ReportBackupCompletionStatus(ObjectStorageNativeBackupStats &backupStatistics)
{
    m_dataSize = 0;
    if (backupStatistics.m_noOfFilesFailed != 0) {
        ReportJobDetailsWithLabel(SubJobStatus::FAILED,
            OBSPlugin::PROGRESS100,
            "object_storage_plugin_restore_data_completed_with_warn_label",
            JobLogLevel::TASK_LOG_WARNING,
            std::to_string(backupStatistics.m_noOfFilesCopied),
            FormatCapacity(backupStatistics.m_noOfBytesCopied),
            std::to_string(backupStatistics.m_noOfFilesFailed));
    } else {
        ReportJobDetailsWithLabel(SubJobStatus::COMPLETED,
            OBSPlugin::PROGRESS100,
            "object_storage_plugin_restore_data_completed_label",
            JobLogLevel::TASK_LOG_INFO,
            std::to_string(backupStatistics.m_noOfFilesCopied),
            FormatCapacity(backupStatistics.m_noOfBytesCopied));
    }
    m_dataSize = backupStatistics.m_noOfBytesCopied/NUMBER1024;
    return true;
}

bool ObjectStorageRestoreJob::UpdateBackupStartTimeInSharedResource(ObjectStorageBackupSubJob& restoreSubJob)
{
    INFOLOG("Enter UpdateBackupStartTimeInSharedResource, %d, %d, %d",
        m_generalInfo.m_restoreCopyIndex, m_generalInfo.m_restoreDeleteIndex, restoreSubJob.m_SubTaskType);
    if (!IsUpdateBackupStartTimeRequired(restoreSubJob, m_generalInfo)) {
        return true;
    }
    ObjectStorageNativeGeneral generalInfo {};
    if (!LockGeneralResource(m_jobId)) {
        ERRLOG("LockGeneralResource failed, jobid:%s", m_jobId.c_str());
        return false;
    }
    if (!GetGeneralResource(m_jobId, generalInfo)) {
        ERRLOG("GetGeneralResource failed, jobid:%s", m_jobId.c_str());
        UnlockGeneralResource(m_jobId);
        return false;
    }
    if (restoreSubJob.m_SubTaskType == OBSPlugin::SUBJOB_TYPE_DATACOPY_COPY_PHASE) {
        if (!UpdateGeneralInfoForCopyPhase(restoreSubJob, generalInfo)) {
            ERRLOG("UpdateGeneralInfoForCopyPhase failed");
            UnlockGeneralResource(m_jobId);
            return false;
        }
    }
    if (restoreSubJob.m_SubTaskType == OBSPlugin::SUBJOB_TYPE_DATACOPY_DELETE_PHASE) {
        if (!UpdateGeneralInfoForDeletePhase(restoreSubJob, generalInfo)) {
            ERRLOG("UpdateGeneralInfoForDeletePhase failed");
            UnlockGeneralResource(m_jobId);
            return false;
        }
    }
    if (!UpdateGeneralResource(m_jobId, generalInfo)) {
        ERRLOG("UpdateGeneralResource failed");
        UnlockGeneralResource(m_jobId);
        return false;
    }
    if (!UnlockGeneralResource(m_jobId)) {
        ERRLOG("UnlockGeneralResource failed");
        return false;
    }
    return true;
}

bool ObjectStorageRestoreJob::UpdateGeneralInfoForCopyPhase(ObjectStorageBackupSubJob& restoreSubJob,
    ObjectStorageNativeGeneral& generalInfo)
{
    if (generalInfo.m_backupCopyPhaseStartTime == 0) {
        m_generalInfo.m_backupCopyPhaseStartTime = GetCurrentTimeInSeconds();
        if (m_generalInfo.m_backupCopyPhaseStartTime == 0) {
            HCP_Log(ERR, MODULE) << "Get current time of first backup copy subtask failed" << HCPENDLOG;
            return false;
        }
        generalInfo.m_backupCopyPhaseStartTime = m_generalInfo.m_backupCopyPhaseStartTime;
    }

    return true;
}

bool ObjectStorageRestoreJob::UpdateGeneralInfoForDeletePhase(ObjectStorageBackupSubJob& restoreSubJob,
    ObjectStorageNativeGeneral& generalInfo)
{
    if (generalInfo.m_backupDelPhaseStartTime == 0) {
        m_generalInfo.m_backupDelPhaseStartTime = GetCurrentTimeInSeconds();
        if (m_generalInfo.m_backupDelPhaseStartTime == 0) {
            HCP_Log(ERR, MODULE) << "Get current time of first backup del subtask failed" << HCPENDLOG;
            return false;
        }
        generalInfo.m_backupDelPhaseStartTime = m_generalInfo.m_backupDelPhaseStartTime;
    }
    return true;
}

bool ObjectStorageRestoreJob::InitJobInfo()
{
    if (GetAuthExtendInfo() != Module::SUCCESS) {
        return false;
    }
    if (!GetRepoPaths()) {
        return false;
    }
    if (!Module::JsonHelper::JsonStringToStruct(m_restoreJobPtr->extendInfo, m_dataLayoutExt)) {
        HCP_Log(ERR, MODULE) << "init (datalayout info) job failed" << HCPENDLOG;
        return false;
    }
    if (!Module::JsonHelper::JsonStringToStruct(m_restoreJobPtr->targetEnv.extendInfo, m_targetEnvExt)) {
        HCP_Log(ERR, MODULE) << "init (targetEnv info) job failed" << HCPENDLOG;
        return false;
    }
    HCP_Log(INFO, MODULE) << "extend info job" << m_restoreJobPtr->extendInfo
                          << "storage type:" << m_targetEnvExt.storageType << HCPENDLOG;

    return true;
}

bool ObjectStorageRestoreJob::ParseCopyExtendInfo(int copyIndex, BackupObjectInfosInCopy& backupObjectCopyInfo)
{
    AppProtect::CopyDataType copyDataType = m_restoreJobPtr->copies[copyIndex].dataType;
    if (copyDataType == AppProtect::CopyDataType::CLOUD_STORAGE_COPY
        || copyDataType == AppProtect::CopyDataType::TAPE_STORAGE_COPY) {
        Json::Value extendInfoJsonVal;
        if (!Module::JsonHelper::JsonStringToJsonValue(m_restoreJobPtr->copies[copyIndex].extendInfo,
            extendInfoJsonVal)) {
            ERRLOG("Convert to extendInfoJsonVal failed.");
            return false;
        }
        if (!(extendInfoJsonVal.isObject() && extendInfoJsonVal.isMember("extendInfo"))) {
            ERRLOG("extendInfo is emtpy.");
            return false;
        }
        Json::Value archvieExtValue = extendInfoJsonVal["extendInfo"];
        if (!Module::JsonHelper::JsonValueToStruct(archvieExtValue, backupObjectCopyInfo)) {
            ERRLOG("JsonStringToStruct failed, archvieExtValue");
            return false;
        }
        return true;
    }
    std::string extendInfo = m_restoreJobPtr->copies[copyIndex].extendInfo;
    if (!Module::JsonHelper::JsonStringToStruct(extendInfo, backupObjectCopyInfo)) {
        ERRLOG("get copy failed");
        return false;
    }
    return true;
}

bool ObjectStorageRestoreJob::ProcessFsRelatons(BackupObjectInfosInCopy& backupObjectCopyInfo)
{
    const std::vector<FsRelation>& relations = backupObjectCopyInfo.fsRelations.relations;
    if (relations.empty()) {
        return true;
    }
    std::unordered_map<std::string, std::string> oldAndNewESNMap;
    for (auto& fsRelation : relations) {
        oldAndNewESNMap[fsRelation.oldEsn] = fsRelation.newEsn;
    }
    for (auto& objectInfo : backupObjectCopyInfo.objectInfos) {
        if (oldAndNewESNMap.count(objectInfo.esn) == 0) {
            ERRLOG("fsRelation exception,oldEsn=%s", objectInfo.esn.c_str());
            return false;
        }
        objectInfo.esn = oldAndNewESNMap[objectInfo.esn];
    }
    return true;
}

bool ObjectStorageRestoreJob::InitRepoPathsByCopyIndex(int copyIndex)
{
    BackupObjectInfosInCopy backupObjectCopyInfo;
    if (!ParseCopyExtendInfo(copyIndex, backupObjectCopyInfo)) {
        ERRLOG("parseCopyExtendInfo failed");
        return false;
    }
    if (!ProcessFsRelatons(backupObjectCopyInfo)) {
        ERRLOG("ProcessFsRelatons failed,copyIndex=%d", copyIndex);
        return false;
    }
    auto copyId = m_restoreJobPtr->copies[copyIndex].id;
    m_restoreCopyIdList.emplace_back(copyId);
    for (auto& objectInfo : backupObjectCopyInfo.objectInfos) {
        m_objectInfoMap[copyId][objectInfo.esn] = objectInfo;
        if (!GetRepoPathsWithEsn(copyIndex, objectInfo.esn, backupObjectCopyInfo.pathSuffix)) {
            return false;
        }
    }
    return true;
}

bool ObjectStorageRestoreJob::GetRepoPathsWithEsn(int copyIndex, const std::string& esn,
    const std::string& pathSuffix)
{
    uint32_t repoSize = m_restoreJobPtr->copies[copyIndex].repositories.size();
    auto copyId = m_restoreJobPtr->copies[copyIndex].id;
    // 对于聚合格式, 多个副本共用一个cache仓, agent只填充最后一个副本的路径
    int tempCopyIndex = IsAggregate() ? m_restoreJobPtr->copies.size() - 1 : copyIndex;
    for (uint32_t i = 0; i < repoSize; ++i) {
        auto& repo = m_restoreJobPtr->copies[tempCopyIndex].repositories[i];
        auto repoType = m_restoreJobPtr->copies[tempCopyIndex].repositories[i].repositoryType;
        std::string dataFsEsn;
        if (!GetRepoEsnByExtendInfo(repo.extendInfo, dataFsEsn)) {
            return false;
        }
        if (esn != dataFsEsn) {
            continue;
        }
        if (repo.path.size() <= 0) {
            ERRLOG("repo path is empty,repoType:%d", (int)repoType);
            return false;
        }
        uint32_t pathIndex = m_nodeLevelTaskInfo.GetSubTasksCount() % repo.path.size();
        string copyPath = IsAggregate() ? repo.path[pathIndex]+ '/' + pathSuffix
            : repo.path[pathIndex];
        if (repoType == RepositoryDataType::DATA_REPOSITORY) {
            m_dataFsPathMap[copyId][esn] = copyPath;
            continue;
        }
        if (repoType == RepositoryDataType::META_REPOSITORY) {
            m_metaFsPathMap[copyId][esn] = copyPath;
            continue;
        }
        if (repoType == RepositoryDataType::CACHE_REPOSITORY) { // 只有最后一个副本填充Cache
            m_cacheFs = repo;
        }
    }
    return true;
}

bool ObjectStorageRestoreJob::GetRepoEsnByExtendInfo(const std::string& exetndInfo, std::string& esn)
{
    Json::Value extendInfoJsonVal;
    if (!Module::JsonHelper::JsonStringToJsonValue(exetndInfo, extendInfoJsonVal)) {
        ERRLOG("repo extendInfo is empty");
        return false;
    }
    if (!(extendInfoJsonVal.isObject()
        && extendInfoJsonVal.isMember("esn")
        && extendInfoJsonVal["esn"].isString())) {
        ERRLOG("repo extendInfo has no esn field");
        return false;
    }
    esn = extendInfoJsonVal["esn"].asString();
    if (esn.empty()) {
        ERRLOG("repo extendInfo with esn empty");
        return false;
    }
    return true;
}

void ObjectStorageRestoreJob::PrintCopyPathInfo(const RepositoryDataType::type& repoType)
{
    std::map<std::string, std::unordered_map<std::string, std::string>> *pathMap;
    if (repoType == RepositoryDataType::type::DATA_REPOSITORY) {
        pathMap = &m_dataFsPathMap;
        INFOLOG("copy data pathInfo:");
    } else {
        pathMap = &m_metaFsPathMap;
        INFOLOG("copy meta pathInfo:");
    }
    for (auto it : *pathMap) {
        std::string pathInfo;
        for (auto iter : it.second) {
            pathInfo = pathInfo + iter.first + "," + iter.second + "\n";
        }
        pathInfo = it.first + "," + pathInfo;
        INFOLOG("path:%s", pathInfo.c_str());
    }
}

uint32_t ObjectStorageRestoreJob::GetCopyRestoreOrder(const std::string& copyId)
{
    uint32_t copyRestoreOrder = 0;
    for (uint32_t i = 0; i < m_restoreCopyIdList.size(); i++) {
        if (m_restoreCopyIdList[i] == copyId) {
            copyRestoreOrder = i;
            break;
        }
    }
    return copyRestoreOrder;
}

bool ObjectStorageRestoreJob::GetRepoPaths()
{
    /* MetaFs and BackupFs to be used */
    if (m_restoreJobPtr->copies.empty()) {
        ERRLOG("no copy found in restore job");
        return false;
    }
    if (!ParseCopyExtendInfo(0, m_backupObjectCopyInfo)) {
        ERRLOG("parse copy extendInfo failed");
        return false;
    }

    if (IsAggregate()) {
        m_aggregateRestore = true;
        DBGLOG("is aggregate");
    }
    uint32_t copySize = m_restoreJobPtr->copies.size();
    for (uint32_t copyIndex = 0; copyIndex < copySize; copyIndex++) {
        if (!InitRepoPathsByCopyIndex(copyIndex)) {
            return false;
        }
    }
    if (m_cacheFs.path.size() == 0 || m_dataFsPathMap.size() == 0 || m_metaFsPathMap.size() == 0) {
        ERRLOG("cacheFs list size = %u, dataFs list size = %u, metaFs list size = %u",
            m_cacheFs.path.size(), m_dataFsPathMap.size(), m_metaFsPathMap.size());
        return false;
    }
    m_cacheFsPath = m_cacheFs.path[m_nodeLevelTaskInfo.GetSubTasksCount() % m_cacheFs.path.size()];
    DBGLOG("m_cacheFs.remotePath = %s, raw m_cacheFsPath = %s", m_cacheFs.remotePath.c_str(), m_cacheFsPath.c_str());
    PrintCopyPathInfo(RepositoryDataType::type::DATA_REPOSITORY);
    PrintCopyPathInfo(RepositoryDataType::type::META_REPOSITORY);
    m_certFilePath = OBSPlugin::PathJoin(m_cacheFsPath, "cert");
    return true;
}

bool ObjectStorageRestoreJob::PrintJobInfo()
{
    HCP_Log(INFO, MODULE) << "jobPhase: " << m_jobCtrlPhase << HCPENDLOG;
    HCP_Log(INFO, MODULE) << "jobId: " << m_jobId << HCPENDLOG;
    HCP_Log(INFO, MODULE) << "subjobId: " << m_subJobId << HCPENDLOG;
    return true;
}

bool ObjectStorageRestoreJob::SetupMetaFsForRestoreJob()
{
    if (!CreateDirectory(OBSPlugin::PathJoin(m_cacheFsPath, "obs-statistics"))) {
        HCP_Log(ERR, MODULE) << "setup meta-fs for restore job failed" << HCPENDLOG;
        return false;
    }
    if (!CreateDirectory(OBSPlugin::PathJoin(m_cacheFsPath, "scan-statistics"))) {
        HCP_Log(ERR, MODULE) << "setup scan statistics for restore job failed" << HCPENDLOG;
        return false;
    }
    return true;
}

bool ObjectStorageRestoreJob::StartScanner()
{
    ScanConfig scanConfig {};
    FillScanConfig(scanConfig);
    AddFilterRule(scanConfig);

    m_isScannerRestarted = false;
    if (m_scanStats.m_scanStarted && scanConfig.scanCheckPointEnable) {
        m_isScannerRestarted = true;
    }

    m_scanner = ScanMgr::CreateScanInst(scanConfig);
    if (!m_scanner) {
        HCP_Log(ERR, MODULE) << "Start scanner instance failed!" << HCPENDLOG;
        return false;
    }
    if (!m_isScannerRestarted) {
        m_scanner->Enqueue(".");
    }
    if (SCANNER_STATUS::SUCCESS != m_scanner->Start()) {
        HCP_Log(ERR, MODULE) << "Start scanner instance failed" << HCPENDLOG;
        return false;
    }
    m_scanStats.m_scanStarted = true;
    return true;
}

bool ObjectStorageRestoreJob::InitPathForRestore()
{
    HCP_Log(INFO, MODULE) << "Enter InitPathForRestore, copyIndex:" << m_copyIndex << HCPENDLOG;
    m_dcaheAndFcachePath = OBSPlugin::PathJoin(m_cacheFsPath, "restore-job", m_jobId,
        to_string(m_copyIndex), "scan", "meta", m_subfolder);
    if (m_aggregateRestore && m_copyIndex > 0) {
        m_preDcaheAndFcachePath = OBSPlugin::PathJoin(m_cacheFsPath, "restore-job", m_jobId,
            to_string(m_copyIndex - 1), "scan", "meta", m_subfolder);
    }
    m_scanControlFilePath = OBSPlugin::PathJoin(m_cacheFsPath, "restore-job", m_jobId,
        to_string(m_copyIndex), "scan", "control", m_subfolder);
    m_restoreControlFilePath = OBSPlugin::PathJoin(m_cacheFsPath, "restore-job", m_jobId,
        to_string(m_copyIndex), "restore", "control", m_subfolder);
    m_checkPointPath = OBSPlugin::PathJoin(m_cacheFsPath, "restore-job", m_jobId, "restore", "checkpoint", m_subfolder);
    HCP_Log(INFO, MODULE) << "m_dcaheAndFcachePath: " << m_dcaheAndFcachePath << HCPENDLOG;
    HCP_Log(INFO, MODULE) << "m_scanControlFilePath: " << m_scanControlFilePath << HCPENDLOG;
    HCP_Log(INFO, MODULE) << "m_restoreControlFilePath: " << m_restoreControlFilePath << HCPENDLOG;
    HCP_Log(INFO, MODULE) << "m_checkPointPath: " << m_checkPointPath << HCPENDLOG;

    if (!OBSPlugin::RecurseCreateDirectory(m_dcaheAndFcachePath)) {
        return false;
    }
    if (!OBSPlugin::RecurseCreateDirectory(m_scanControlFilePath)) {
        return false;
    }
    if (!OBSPlugin::RecurseCreateDirectory(m_restoreControlFilePath)) {
        return false;
    }
    if (!OBSPlugin::RecurseCreateDirectory(m_checkPointPath)) {
        return false;
    }
    return true;
}

bool ObjectStorageRestoreJob::UnzipDcachefiles(const std::string& metaPath, const std::string& subfolder,
    const std::string& dcaheAndFcachePath)
{
    DBGLOG("Unzip meta files, meta path: %s", metaPath.c_str());
    std::string metaFile = OBSPlugin::PathJoin(metaPath, "filemeta", subfolder, "meta_file_0");
    if (OBSPlugin::IsFileExist(metaFile)) {
        std::string fileMetaPath = OBSPlugin::PathJoin(metaPath, "filemeta", subfolder);
        bool isSuccess = CopyUnCompressMeta(fileMetaPath, dcaheAndFcachePath) == Module::SUCCESS ? true : false;
        return isSuccess;
    }

    string dirCacheZipFileName = OBSPlugin::PathJoin(metaPath, "filemeta", subfolder, "metafile_DIRCACHE.gz");
    string fCacheZipFileName = OBSPlugin::PathJoin(metaPath, "filemeta", subfolder, "metafile_FILECACHE.gz");
    string metaZipFilename = OBSPlugin::PathJoin(metaPath, "filemeta", subfolder, "metafile_META.gz");
    string xmetaZipFileName = OBSPlugin::PathJoin(metaPath, "filemeta", subfolder, "metafile_XMETA.gz");
    if (!OBSPlugin::IsFileExist(dirCacheZipFileName) || !OBSPlugin::IsFileExist(fCacheZipFileName) ||
        !OBSPlugin::IsFileExist(metaZipFilename) || !OBSPlugin::IsFileExist(xmetaZipFileName)) {
        HCP_Log(ERR, MODULE) << "cache file not exist " << HCPENDLOG;
        return false;
    }

    std::vector<std::string> paramList {};
    std::string zipDirCacheFilesCmd = "tar -I pigz -xf " + dirCacheZipFileName + " -C " + dcaheAndFcachePath;
    HCP_Log(INFO, MODULE) << "Run unzip DirCacheFiles cmd : " << zipDirCacheFilesCmd << HCPENDLOG;
    int ret = OBSPlugin::RunShellCmd(zipDirCacheFilesCmd, paramList);
    if (ret != Module::SUCCESS) {
        HCP_Log(ERR, MODULE) << "Run unzip DirCacheFiles cmd failed." << HCPENDLOG;
        return false;
    }

    std::string zipFCacheFilesCmd = "tar -I pigz -xf " + fCacheZipFileName + " -C " + dcaheAndFcachePath;
    HCP_Log(INFO, MODULE) << "Run unzip FCacheFiles cmd : " << zipFCacheFilesCmd << HCPENDLOG;
    ret = OBSPlugin::RunShellCmd(zipFCacheFilesCmd, paramList);
    if (ret != Module::SUCCESS) {
        HCP_Log(ERR, MODULE) << "Run unzip FCacheFiles cmd failed." << HCPENDLOG;
        return false;
    }

    std::string zipMetaFilesCmd = "tar -I pigz -xf " + metaZipFilename + " -C " + dcaheAndFcachePath;
    HCP_Log(INFO, MODULE) << "Run unzip MetaFiles cmd : " << zipMetaFilesCmd << HCPENDLOG;
    ret = OBSPlugin::RunShellCmd(zipMetaFilesCmd, paramList);
    if (ret != Module::SUCCESS) {
        HCP_Log(ERR, MODULE) << "Run unzip MetaFiles cmd failed." << HCPENDLOG;
        return false;
    }

    std::string zipXMetaFilesCmd = "tar -I pigz -xf " + xmetaZipFileName + " -C " + dcaheAndFcachePath;
    HCP_Log(INFO, MODULE) << "Run unzip XMetaFiles cmd : " << zipXMetaFilesCmd << HCPENDLOG;
    ret = OBSPlugin::RunShellCmd(zipXMetaFilesCmd, paramList);
    if (ret != Module::SUCCESS) {
        HCP_Log(ERR, MODULE) << "Run unzip XMetaFiles cmd failed." << HCPENDLOG;
        return false;
    }

    HCP_Log(INFO, MODULE) << "untar metafile.tar.gz success." << HCPENDLOG;
    return true;
}

void ObjectStorageRestoreJob::FillScanConfigBasedOnEnviroment(ScanConfig& scanConfig)
{
    scanConfig.writeQueueSize = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_SCAN_X8000_WRITE_QUEUE_SIZE");
    scanConfig.scanMetaFileSize = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_SCAN_X8000_DEFAULT_META_FILE_SIZE");
    scanConfig.producerThreadCount = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_SCAN_X8000_PRODUCER_THREAD_COUNT");

    std::string deployType = OBSPlugin::GetDeployType();
    if (deployType == X3000_DEPLOY_TYPE) {
        scanConfig.writeQueueSize = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
            "DME_OBS_SCAN_X3000_WRITE_QUEUE_SIZE");
        scanConfig.scanMetaFileSize = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
            "DME_OBS_SCAN_X3000_DEFAULT_META_FILE_SIZE");
        scanConfig.producerThreadCount = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
            "DME_OBS_SCAN_X3000_PRODUCER_THREAD_COUNT");

        HCP_Log(INFO, MODULE) << "Use x3000 config."
            << " writeQueueSize - " << scanConfig.writeQueueSize
            << " scanMetaFileSize - " << scanConfig.scanMetaFileSize
            << " producerThreadCount - " << scanConfig.producerThreadCount
            << HCPENDLOG;
    } else if (deployType == X6000_DEPLOY_TYPE) {
        scanConfig.writeQueueSize = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
            "DME_OBS_SCAN_X3000_WRITE_QUEUE_SIZE");
        scanConfig.scanMetaFileSize = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
            "DME_OBS_SCAN_X3000_DEFAULT_META_FILE_SIZE");
        scanConfig.producerThreadCount = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
            "DME_OBS_SCAN_X3000_PRODUCER_THREAD_COUNT");
        HCP_Log(INFO, MODULE) << "Use x6000 config." << HCPENDLOG;
    }

    if (m_authInfo.getStorageType() == StorageType::ALI) {
        scanConfig.producerThreadCount = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
            "DME_OBS_SCAN_X8000_PRODUCER_THREAD_COUNT");
    }
}

void ObjectStorageRestoreJob::FillScanConfig(ScanConfig &scanConfig)
{
    HCP_Log(DEBUG, MODULE) << "Begin to fill scan config." << HCPENDLOG;
    scanConfig.jobId = m_jobId;
    scanConfig.reqID = OBSPlugin::GenerateHash(m_jobId);

    /* config meta path */
    scanConfig.curDcachePath = m_dcaheAndFcachePath;
    scanConfig.prevDcachePath = m_preDcaheAndFcachePath;
    scanConfig.metaPathForCtrlFiles = m_scanControlFilePath;
    scanConfig.metaPath = m_dcaheAndFcachePath;
    HCP_Log(INFO, MODULE) << "meta path: " << scanConfig.metaPath + ", "
                          << "control path: " << scanConfig.metaPathForCtrlFiles << HCPENDLOG;
    scanConfig.scanType = ScanJobType::CONTROL_GEN;
    scanConfig.scanIO = IOEngine::DEFAULT;
    scanConfig.generatorIsFull = (m_aggregateRestore && m_copyIndex != 0) ? false : true;
    scanConfig.failureRecordRootPath = m_failureRecordRoot;

    /* thraed num */
    scanConfig.maxCommonServiceInstance = 1;
    scanConfig.usrData = (void *)this;

    /* Callbacks Regiter */
    scanConfig.scanResultCb = ScannerCtrlFileCallBack;
    scanConfig.deleteCtrlCb = RestoreDelCtrlCallBack;

    scanConfig.triggerTime = GetCurrentTimeInSeconds();

    // fill
    FillScanConfigForNative(scanConfig);
    FillScanConfigBasedOnEnviroment(scanConfig);

    HCP_Log(DEBUG, MODULE) << "End to fill scan config." << HCPENDLOG;
}

void ObjectStorageRestoreJob::AddFilterRule(ScanConfig& scanConfig)
{
    HCP_Log(DEBUG, MODULE) << "Enter AddFilterRule" << HCPENDLOG;
    // not fine grained ,return
    if (!m_fineGrainedRestore) {
        return;
    }
    vector<string> prefixFilterRule;
    vector<string> fileFilterRule;
    for (size_t i = 0; i < m_restoreJobPtr->restoreSubObjects.size(); ++i) {
        string path = m_restoreJobPtr->restoreSubObjects[i].name;
        // is prefix or object , the last char is not '/' is object
        if (path.back() != SLASH_CH) { // object
            HCP_Log(INFO, MODULE) << "object filter rule: " << path << HCPENDLOG;
            fileFilterRule.push_back(path);
        } else { // prefix
            HCP_Log(INFO, MODULE) << "prefix filter rule: " << path << HCPENDLOG;
            prefixFilterRule.push_back(path);
        }
    }
    scanConfig.dCtrlFltr = prefixFilterRule;
    scanConfig.fCtrlFltr = fileFilterRule;
    return;
}

bool ObjectStorageRestoreJob::StartRestore(ObjectStorageBackupSubJob restoreSubJob)
{
    BackupParams backupParams {};
    FillRestoreConfig(backupParams, restoreSubJob);

    m_backup = FS_Backup::BackupMgr::CreateBackupInst(backupParams);
    if (m_backup == nullptr) {
        HCP_Log(ERR, MODULE) << "Create backup instance failed" << HCPENDLOG;
        return false;
    }
    if (BackupRetCode::SUCCESS != m_backup->Enqueue(restoreSubJob.m_ControlFile)) {
        HCP_Log(ERR, MODULE) << "enqueue backup instance failed" << HCPENDLOG;
        return false;
    }
    if (BackupRetCode::SUCCESS != m_backup->Start()) {
        HCP_Log(ERR, MODULE) << "Start backup instance failed" << HCPENDLOG;
        return false;
    }
    return true;
}

void ObjectStorageRestoreJob::FillBackupReplacePolicy(BackupParams &backupParams)
{
    if (m_dataLayoutExt.fileReplaceStrategy == OBSPlugin::REPLACE_POLICY_REPLACE_EXSISTING_FILESNFOLDERS) {
        backupParams.commonParams.restoreReplacePolicy = RestoreReplacePolicy::OVERWRITE;
    } else if (m_dataLayoutExt.fileReplaceStrategy == OBSPlugin::REPLACE_POLICY_IGNORE_EXSISTNG_FILES) {
        backupParams.commonParams.restoreReplacePolicy = RestoreReplacePolicy::IGNORE_EXIST;
    } else if (m_dataLayoutExt.fileReplaceStrategy == OBSPlugin::REPLACE_POLICY_REPLACE_ONLY_IF_OLDER) {
        backupParams.commonParams.restoreReplacePolicy = RestoreReplacePolicy::OVERWRITE_OLDER;
    } else {
        backupParams.commonParams.restoreReplacePolicy = RestoreReplacePolicy::NONE;
    }
}

void ObjectStorageRestoreJob::FillBackupCommonParams(BackupParams &backupParams)
{
    FillBackupBasedCommonParams(backupParams);
    FillBackupReplacePolicy(backupParams);
    HCP_Log(INFO, MODULE) <<"Backup params meta path:" << backupParams.commonParams.metaPath << HCPENDLOG;

    if (m_aggregateRestore) {
        HCP_Log(INFO, MODULE) <<"FillBackupCommonParams:aggregate:yes" << HCPENDLOG;
        backupParams.commonParams.backupDataFormat = BackupDataFormat::AGGREGATE;
        backupParams.commonParams.maxAggregateFileSize =
            std::stoul(m_backupObjectCopyInfo.advParms.m_maxSizeAfterAggregate);
        backupParams.commonParams.maxFileSizeToAggregate =
            std::stoul(m_backupObjectCopyInfo.advParms.m_maxSizeToAggregate);
        int aggregateThreadNum =
            GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION, "DME_OBS_AGGR_THREAD_POOL_CNT");

        HCP_Log(DEBUG, MODULE) <<" FillBackupCommonParams Aggregate Info, aggregateThreadNum: "<< aggregateThreadNum <<
            " maxAggregateFileSize: " << backupParams.commonParams.maxAggregateFileSize <<
            " maxFileSizeToAggregate: " << backupParams.commonParams.maxFileSizeToAggregate << HCPENDLOG;
        if (aggregateThreadNum > MAX_AGGREGATE_NUM || aggregateThreadNum <= 0) {
            aggregateThreadNum = DEFAULT_AGGREGATE_NUM;
        }
        backupParams.commonParams.aggregateThreadNum = aggregateThreadNum;
        backupParams.commonParams.genSqlite = false;
    } else {
        HCP_Log(INFO, MODULE) <<"FillBackupCommonParams:native:yes" << HCPENDLOG;
        backupParams.commonParams.backupDataFormat = BackupDataFormat::NATIVE;
        backupParams.commonParams.genSqlite = true;
    }
}

void ObjectStorageRestoreJob::FillBackupBasedCommonParams(BackupParams &backupParams)
{
    backupParams.commonParams.jobId = m_jobId;
    backupParams.commonParams.subJobId = m_subJobId;
    backupParams.commonParams.reqID = m_subJobRequestId;
    if (m_backupObjectCopyInfo.advParms.m_isBackupAcl == "true") {
        HCP_Log(INFO, MODULE) <<"backup acl" << HCPENDLOG;
        backupParams.commonParams.writeAcl = true;
    }
    backupParams.commonParams.useSubJobSqlite = true;
    backupParams.commonParams.writeExtendAttribute = true;
    backupParams.commonParams.failureRecordRootPath = m_failureRecordRoot;
    if (m_authInfo.getStorageType() == StorageType::ALI) {
        backupParams.commonParams.maxBufferCnt = GetConfigIntKey(
            OBSPlugin::DME_OBS_CONFIG_SECTION, "DME_OBS_BACKUP_MAX_ALI_QPS_RW");
    } else {
        backupParams.commonParams.maxBufferCnt = GetConfigIntKey(
            OBSPlugin::DME_OBS_CONFIG_SECTION, "DME_OBS_BACKUP_MAX_OBS_QPS_PUT");
    }
    backupParams.commonParams.sqliteLocalPath = GetConfigStrKey(DME_OBS_CONFIG_SECTION,
        "OBS_SQLITE_DIR");
    backupParams.commonParams.maxBufferSize = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_BACKUP_MAX_TOTAL_BLOCK_BUFFER_SIZE");
    backupParams.commonParams.blockSize = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_BACKUP_BLOCK_SIZE");
    backupParams.commonParams.maxBlockNum = GetConfigIntKey(OBSPlugin::DME_OBS_CONFIG_SECTION,
        "DME_OBS_BACKUP_BLOCK_NUM");
}

void ObjectStorageRestoreJob::FillBackupConfigPhase(
    BackupParams &backupParams, ObjectStorageBackupSubJob &restoreSubJob)
{
    if (restoreSubJob.m_SubTaskType == OBSPlugin::SUBJOB_TYPE_DATACOPY_COPY_PHASE) {
        backupParams.phase = BackupPhase::COPY_STAGE;
    } else if (restoreSubJob.m_SubTaskType == OBSPlugin::SUBJOB_TYPE_DATACOPY_DELETE_PHASE) {
        backupParams.phase = BackupPhase::DELETE_STAGE;
    }
}

void ObjectStorageRestoreJob::FillRestoreConfig(BackupParams &backupParams, ObjectStorageBackupSubJob &restoreSubJob)
{
    backupParams.srcEngine = BackupIOEngine::POSIX;
    backupParams.dstEngine = BackupIOEngine::OBJECTSTORAGE;
    backupParams.backupType = BackupType::RESTORE;
    FillBackupConfigPhase(backupParams, restoreSubJob);
    FillRestoreAdvanceParams(backupParams, restoreSubJob);
    FillBackupCommonParams(backupParams);
    if (!IsGrainRestoreOrPermanentIncrCopy()) {
        backupParams.commonParams.controlFile = OBSPlugin::GetFileName(restoreSubJob.m_ControlFile);
    }
    backupParams.scanAdvParams.metaFilePath = OBSPlugin::PathJoin(m_cacheFsPath, restoreSubJob.dcacheAndFcachePath);
    string metaFsPath = m_metaFsPathMap[restoreSubJob.copyId][restoreSubJob.esn];
    backupParams.commonParams.metaPath = OBSPlugin::PathJoin(metaFsPath, restoreSubJob.metaFilePath);

    DBGLOG("FillRestoreConfig, metaFilePath: %s, metaPath: %s",
        backupParams.scanAdvParams.metaFilePath.c_str(),
        backupParams.commonParams.metaPath.c_str());
}

void ObjectStorageRestoreJob::FillRestoreAdvanceParams(
    BackupParams &backupParams, ObjectStorageBackupSubJob &restoreSubJob)
{
    HostBackupAdvanceParams srcAdvanceParams {};
    ObjectBackupAdvanceParams dstAdvanceParams {};
    string dataFsPath = m_dataFsPathMap[restoreSubJob.copyId][restoreSubJob.esn];
    srcAdvanceParams.dataPath = OBSPlugin::PathJoin(dataFsPath, restoreSubJob.dataCachePath);
    srcAdvanceParams.threadNum = GetConfigIntKey(DME_OBS_CONFIG_SECTION,
        "DME_OBS_BACKUP_SERVER_THREAD_NUM");
    srcAdvanceParams.maxMemory = GetConfigIntKey(DME_OBS_CONFIG_SECTION,
        "DME_OBS_BACKUP_SERVER_MAX_MEMORY_SIZE");
    dstAdvanceParams.threadNum = GetConfigIntKey(DME_OBS_CONFIG_SECTION,
        "DME_OBS_BACKUP_SERVER_THREAD_NUM");
    dstAdvanceParams.maxMemory = GetConfigIntKey(DME_OBS_CONFIG_SECTION,
        "DME_OBS_BACKUP_SERVER_MAX_MEMORY_SIZE");
    if (!m_dataLayoutExt.bucketName.empty()) {
        dstAdvanceParams.dstBucket.bucketName = m_dataLayoutExt.bucketName;
    }
    if (!m_dataLayoutExt.prefix.empty()) {
        dstAdvanceParams.dstBucket.prefix.emplace_back(m_dataLayoutExt.prefix);
    }
    dstAdvanceParams.dstBucket.delimiter = "/";

    dstAdvanceParams.dataPath = OBSPlugin::PathJoin(dataFsPath, restoreSubJob.dataCachePath);
    dstAdvanceParams.cachePath = restoreSubJob.checkpointCachePath;
    dstAdvanceParams.isfineGrainedRestore = m_fineGrainedRestore;
    std::unordered_map<std::string, std::vector<std::string>> bucketPrefixMap; // key: <endpoint,buckets>
    bool encodeSwitch = false;
    ReadEncodeBucket(bucketPrefixMap, encodeSwitch);
    if (encodeSwitch) {
        dstAdvanceParams.dstBucket.encodeEnable = true;
    }
    for (auto& prefixInfo : restoreSubJob.prefixInfo) {
        ObsBucket tmpObsBucket;
        tmpObsBucket.bucketName = prefixInfo.bucketName;
        tmpObsBucket.encodeEnable = CheckEncode(m_authInfo.endPoint, tmpObsBucket.bucketName,
            bucketPrefixMap, encodeSwitch);
        HCP_Log(DEBUG, MODULE) << "bucketName: " << tmpObsBucket.bucketName << HCPENDLOG;
        dstAdvanceParams.buckets.emplace_back(tmpObsBucket);
    }

    FillStorageConfig(dstAdvanceParams.authArgs);
    backupParams.srcAdvParams = make_shared<HostBackupAdvanceParams>(srcAdvanceParams);
    backupParams.dstAdvParams = make_shared<ObjectBackupAdvanceParams>(dstAdvanceParams);
}

void ObjectStorageRestoreJob::FillStorageConfig(Module::StorageConfig& storageConfig)
{
    storageConfig.storageType = static_cast<Module::StorageType>(stoi(m_targetEnvExt.storageType));
    storageConfig.verifyInfo.endPoint = m_authInfo.endPoint;
    storageConfig.verifyInfo.accessKey = m_authInfo.ak;
    storageConfig.verifyInfo.secretKey = m_authInfo.sk;
    if (m_authInfo.openHttps()) {
        std::string certContent = OBSPlugin::Base64Decode(m_authInfo.certification);
        CertManager certMgr(m_certFilePath, m_jobId);
        certMgr.SaveFile(certContent);
        storageConfig.verifyInfo.useHttps = true;
        storageConfig.verifyInfo.caPath = certMgr.GetCAPath();
        storageConfig.verifyInfo.caFile = certMgr.GetCAFile();
        storageConfig.verifyInfo.certHttps = certContent;
    }
    if (m_authInfo.openProxy()) {
        storageConfig.verifyInfo.useProxy = true;
        storageConfig.verifyInfo.proxyHostName = m_authInfo.proxyHostName;
        storageConfig.verifyInfo.proxyUserName = m_authInfo.proxyUserName;
        storageConfig.verifyInfo.proxyUserPwd = m_authInfo.proxyUserPwd;
    }
}

void ObjectStorageRestoreJob::ScannerCtrlFileCallBack(void* usrData, const string &controlFilePath)
{
    HCP_Log(DEBUG, MODULE) << "Callback Received for control File path:"<< WIPE_SENSITIVE(controlFilePath) << HCPENDLOG;
    return;
}

void ObjectStorageRestoreJob::RestoreDelCtrlCallBack(void* usrData, const string &controlFilePath)
{
    HCP_Log(DEBUG, MODULE) << "Callback Received for control File path:"<< WIPE_SENSITIVE(controlFilePath) << HCPENDLOG;
    return;
}

bool ObjectStorageRestoreJob::UpdateBackupStatistics(BackupStats &backupStatistics)
{
    m_dataSize = backupStatistics.noOfBytesCopied / NUMBER1024;
    UpdateBackupSubTaskStatistics(backupStatistics, m_cacheFsPath, 0, 0);
    PrintBackupStatistics(backupStatistics, m_jobId, m_backupStatus, m_subJobId);

    if (!m_nodeLevelTaskInfo.CanSendLogReportToPM(m_jobId)) {
        ReportJobDetailsWithLabel(SubJobStatus::RUNNING, OBSPlugin::PROGRESS0, "", JobLogLevel::TASK_LOG_INFO);
        return true;
    }

    ObjectStorageNativeBackupStats backupStatsResource {};
    bool bReportToPM = false;

    if (!LockBackupStatsResource(m_jobId)) {
        ERRLOG("LockBackupStatsResource failed");
        UnlockBackupStatsResource(m_jobId);
        ReportJobDetailsWithLabel(SubJobStatus::RUNNING, OBSPlugin::PROGRESS0, "", JobLogLevel::TASK_LOG_INFO);
        return false;
    }

    if (!GetBackupStatsResource(m_jobId, backupStatsResource)) {
        ERRLOG("GetBackupStatsResource failed");
        ReportJobDetailsWithLabel(SubJobStatus::RUNNING, OBSPlugin::PROGRESS0, "", JobLogLevel::TASK_LOG_INFO);
        UnlockBackupStatsResource(m_jobId);
        return false;
    }

    if ((GetCurrentTimeInSeconds() - backupStatsResource.m_lastLogReportTime) >
        OBSPlugin::BACKUP_REPORT_CIRCLE_TIME_IN_SEC) {
        bReportToPM = true;
        backupStatsResource.m_lastLogReportTime = GetCurrentTimeInSeconds();
        UpdateBackupStatsResource(m_jobId, backupStatsResource);
    }
    UnlockBackupStatsResource(m_jobId);

    if (!bReportToPM) {
        ReportJobDetailsWithLabel(SubJobStatus::RUNNING, OBSPlugin::PROGRESS0, "", JobLogLevel::TASK_LOG_INFO);
    } else {
        ReportBackupRunningStatus();
    }
    return true;
}

void ObjectStorageRestoreJob::PrintBackupCopyInfo(ObjectStorageNativeBackupStats &backupStatistics)
{
    if (m_jobResult != AppProtect::JobResult::type::SUCCESS) {
        HCP_Log(INFO, MODULE)
            << "\n\n----------------"
            << "\nRestore Job Report"
            << "\n------------------"
            << "\nStatus    : Restore failed"
            << "\nJob Id    : " << m_jobId
            << "\n\n" << HCPENDLOG;
        PrintFinalScannerStats(m_scanStats);
        PrintFinalBackupStats(backupStatistics);
        return;
    }

    time_t jobEndTime = GetCurrentTimeInSeconds();
    std::string jobStartTimeStr = FormatTimeToStr(m_generalInfo.m_jobStartTime);
    std::string jobEndTimeStr = FormatTimeToStr(jobEndTime);
    int32_t jobDuration = double(jobEndTime - m_generalInfo.m_jobStartTime);

    HCP_Log(INFO, MODULE)
        << "\n\n----------------"
        << "\nRestore Job Report"
        << "\n------------------"
        << "\nStatus                    : Restore successful"
        << "\nJob Start time            : " << jobStartTimeStr
        << "\nJob End time              : " << jobEndTimeStr
        << "\nJob Duration (seconds)    : " << jobDuration
        << "\nJob Id                    : " << m_jobId
        << "\n\n" << HCPENDLOG;

    PrintFinalScannerStats(m_scanStats);
    PrintFinalBackupStats(backupStatistics);
    return;
}

void ObjectStorageRestoreJob::KeepPluginAlive()
{
    HCP_Log(INFO, MODULE) << "Enter KeepPluginAlive" << HCPENDLOG;
    ActionResult result;
    SubJobDetails subJobDtls;
    LogDetail logDetail{};
    std::vector<LogDetail> logDetails;
    uint32_t reportCnt = 0;
    while (!m_isAbort && !m_generateSubjobFinish) {
        std::this_thread::sleep_for(std::chrono::seconds(OBSPlugin::REPORT_RUNNING_INTERVAL));
        if (reportCnt % OBSPlugin::REPORT_RUNNING_TIMES == 0) {
            REPORT_LOG2AGENT(subJobDtls, result, logDetails, logDetail, 0, 0, SubJobStatus::RUNNING);
        }
        reportCnt++;
    }
    INFOLOG("Keep Alive thread exit!, %d, %d", m_isAbort.load(), m_generateSubjobFinish.load());
}

int ObjectStorageRestoreJob::GetAuthExtendInfo()
{
    HCP_Log(DEBUG, MODULE) << "Enter GetAuthExtendInfo"<< HCPENDLOG;
    if (!Module::JsonHelper::JsonStringToStruct(m_restoreJobPtr->targetEnv.auth.extendInfo, m_authInfo)) {
        HCP_Log(ERR, MODULE) << "parse GetAuthExtendInfo failed. authExtendInfo is: "
                             << WIPE_SENSITIVE(m_restoreJobPtr->targetEnv.auth.extendInfo)
                             << "jobId is: " << m_restoreJobPtr->jobId << HCPENDLOG;
        return Module::FAILED;
    }
    return Module::SUCCESS;
}

bool ObjectStorageRestoreJob::HandleSingleCopyForAggregate(uint32_t copyIndex)
{
    auto copyId = m_restoreCopyIdList[copyIndex];
    auto& copyMap = m_dataFsPathMap[copyId];
    uint32_t subTaskPrio =
        copyIndex * OBSPlugin::SUBJOB_TYPE_AGGREGATE_BASE_PRIO + OBSPlugin::SUBJOB_TYPE_PREFIX_SCAN_PHASE_PRIO;
    for (auto iter : copyMap) {
        std::string esn = iter.first;
        HCP_Log(INFO, MODULE) << "generate subTask,copyId:" << copyId << ",esn: " << esn << HCPENDLOG;
        if (!CreateSubJobTaskToCreateRestoreSubJobInner(esn, copyId, subTaskPrio)) {
            HCP_Log(ERR, MODULE) << "CreateSubJobTaskToCreateRestoreSubJobInner failed. " << HCPENDLOG;
            return false;
        }
    }
    return true;
}

bool ObjectStorageRestoreJob::CreateSubJobTaskToCreateRestoreSubJob()
{
    if (m_aggregateRestore) { // 目录格式，多副本
        if (m_fineGrainedRestore) {
            // 只需要恢复最后两次副本之间的差异
            HandleSingleCopyForAggregate(m_restoreCopyIdList.size() - 1);
        } else {
            // 依次恢复所有副本
            for (uint32_t copyIndex = 0; copyIndex < m_restoreCopyIdList.size(); copyIndex++) {
                HandleSingleCopyForAggregate(copyIndex);
            }
        }
    } else { // 快照格式（永久增量生成），只有一次副本（单副本），因此只恢复一次副本
        HandleSingleCopyForAggregate(0);
    }
    HCP_Log(INFO, MODULE) << "CreateSubJobTaskToCreateRestoreSubJobInner success. " << HCPENDLOG;
    // 创建扫描任务结束统计子任务
    if (!CreateScanJobTeardownTask()) {
        HCP_Log(ERR, MODULE) << "Create ScanJobTeardown subtask failed" << HCPENDLOG;
        return false;
    }

    if (!CreateRestoreJobTeardownTask()) {
        HCP_Log(ERR, MODULE) << "Generate tear down subJob failed" << HCPENDLOG;
        return false;
    }

    uint32_t subJobPrio = (m_restoreCopyIdList.size() + 1) * OBSPlugin::SUBJOB_TYPE_AGGREGATE_BASE_PRIO
        + OBSPlugin::SUBJOB_TYPE_ZERO_BACKUP_JUDGE_PRIO;
    if (!CreateZeroBackupJudgeTask(subJobPrio)) {
        HCP_Log(WARN, MODULE) << "Generate zero backup judge task failed" << HCPENDLOG;
    }

    return true;
}

bool ObjectStorageRestoreJob::CreateSubJobTaskToCreateRestoreSubJobInner(const std::string& esn,
    const std::string& copyId, int subTaskPrio)
{
    std::string backupSubJobStr {};
    ObjectStorageBackupSubJob backupSubJob {};
    SubJob subJob {};
    backupSubJob.m_SubTaskType = OBSPlugin::SUBJOB_TYPE_PREFIX_SCAN_PHASE;
    backupSubJob.esn = esn;
    backupSubJob.copyId = copyId;
    const auto& objectInfo = m_objectInfoMap[copyId][esn];
    backupSubJob.prefixInfo.assign(objectInfo.prefixInfo.begin(), objectInfo.prefixInfo.end());

    if (!Module::JsonHelper::StructToJsonString(backupSubJob, backupSubJobStr)) {
        HCP_Log(ERR, MODULE) << "parse backupSubJob failed failed" << HCPENDLOG;
        return false;
    }
    subJob.__set_jobId(m_jobId);
    subJob.__set_jobType(SubJobType::BUSINESS_SUB_JOB);
    subJob.__set_policy(ExecutePolicy::ANY_NODE);
    subJob.__set_jobInfo(backupSubJobStr);
    subJob.__set_ignoreFailed(false);

    if (m_idGenerator == nullptr) {
        InitIdGenerator();
    }
    std::string createSubJobName =
        OBSPlugin::SUBJOB_TYPE_PREFIX_SCAN_JOBNAME + std::to_string(m_idGenerator->GenerateId());
    subJob.__set_jobPriority(subTaskPrio);
    subJob.__set_jobName(createSubJobName);

    do {
        if (IsAbortJob()) {
            WARNLOG("Exit received Abort for taskid: %s, subtaskid: %s", m_jobId.c_str(), m_subJobId.c_str());
            break;
        }
        int ret = CreateSubTask(subJob);
        if (ret == Module::SUCCESS) {
            break;
        } else if (ret == Module::RETRY) {
            HCP_Log(WARN, MODULE) << "Create subtask failed with retriable error"<< HCPENDLOG;
            SleepForCreateSubTaskError();
            continue;
        } else {
            ERRLOG("Exit CreateSubJobTaskToCreateRestoreSubJobInner, CreateSubtask failed");
            return false;
        }
    } while (true);
    return true;
}

bool ObjectStorageRestoreJob::CreateScanJobTeardownTask()
{
    std::string jobName = SUBJOB_TYPE_SCAN_TEARDOWN_JOBNAME;
    uint32_t subJobPrio = m_restoreCopyIdList.size() * OBSPlugin::SUBJOB_TYPE_AGGREGATE_BASE_PRIO +
                          SUBJOB_TYPE_SCAN_TEAR_DOWN_PHASE_PRIO;
    return CreateSubTaskWithRetry(SUBJOB_TYPE_SCAN_TEARDOWN_PHASE, jobName, subJobPrio);
}

bool ObjectStorageRestoreJob::CreateRestoreJobTeardownTask()
{
    std::string jobName = OBSPlugin::SUBJOB_TYPE_TEARDOWN_JOBNAME;
    uint32_t subJobPrio = (m_restoreCopyIdList.size() + 1) * OBSPlugin::SUBJOB_TYPE_AGGREGATE_BASE_PRIO
        + OBSPlugin::SUBJOB_TYPE_TEARDOWN_PHASE_PRIO;
    return CreateSubTaskWithRetry(SUBJOB_TYPE_TEARDOWN_PHASE, jobName, subJobPrio);
}

bool ObjectStorageRestoreJob::IsAggregate()
{
    return m_backupObjectCopyInfo.advParms.m_aggregateSwitch == "true";
}

int ObjectStorageRestoreJob::CheckCreateBucketExist(bool& exit)
{
    Module::StorageConfig storageConfig;
    storageConfig.storageType = static_cast<Module::StorageType>(stoi(m_targetEnvExt.storageType));
    storageConfig.verifyInfo.endPoint = m_authInfo.endPoint;
    storageConfig.verifyInfo.accessKey = m_authInfo.ak;
    storageConfig.verifyInfo.secretKey = m_authInfo.sk;
    if (m_authInfo.openHttps()) {
        std::string certContent = OBSPlugin::Base64Decode(m_authInfo.certification);
        CertManager certMgr(m_certFilePath, m_jobId);
        certMgr.SaveFile(certContent);
        storageConfig.verifyInfo.useHttps = true;
        storageConfig.verifyInfo.certHttps = certContent;
        storageConfig.verifyInfo.caPath = certMgr.GetCAPath();
        storageConfig.verifyInfo.caFile = certMgr.GetCAFile();
    }
    if (m_authInfo.openProxy()) {
        storageConfig.verifyInfo.useProxy = true;
        storageConfig.verifyInfo.proxyHostName = m_authInfo.proxyHostName;
        storageConfig.verifyInfo.proxyUserName = m_authInfo.proxyUserName;
        storageConfig.verifyInfo.proxyUserPwd = m_authInfo.proxyUserPwd;
    }
    auto cloudService = Module::CloudServiceManager::CreateInst(storageConfig);
    if (cloudService == nullptr) {
        HCP_Log(ERR, MODULE) << "CreateInst Failed" << HCPENDLOG;
        return Module::FAILED;
    }
    std::unique_ptr<Module::HeadBucketRequest> req = std::make_unique<Module::HeadBucketRequest>();
    std::unique_ptr<Module::HeadBucketResponse> resp = nullptr;
    req->bucketName = m_dataLayoutExt.bucketName;
    Module::OBSResult ret = cloudService->IsBucketExist(req, resp);
    HCP_Log(INFO, MODULE) << "bucket name:" << m_dataLayoutExt.bucketName << HCPENDLOG;
    if (!ret.IsSucc()) {
        HCP_Log(ERR, MODULE) << "Get bucket failed." << HCPENDLOG;
        return Module::FAILED;
    }
    if (resp->isExist) {
        HCP_Log(WARN, MODULE) << "bucket exist." << HCPENDLOG;
        exit = true;
    } else {
        HCP_Log(INFO, MODULE) << "bucket not exist." << HCPENDLOG;
        exit = false;
    }

    return Module::SUCCESS;
}
bool ObjectStorageRestoreJob::UpdateScanStatInfo(const PrefixInfo& prefixInfo,
    ObjectStorageNativeScanStatistics& scanStatistics)
{
    std::string scanStatisticsPath = OBSPlugin::PathJoin(m_cacheFsPath, "scan-statistics");

    UpdateScannerSubTaskStatistics(prefixInfo, scanStatistics, scanStatisticsPath, m_subJobId);
    bool bReportToPM = false;
    if (!UpdateScanStatsResourceWithLock(scanStatistics, bReportToPM)) {
        return false;
    }
    if (!bReportToPM) {
        ReportJobDetails(JobLogLevel::TASK_LOG_INFO, SubJobStatus::RUNNING, OBSPlugin::PROGRESS0);
        return true;
    }
    ObjectStorageNativeScanStatistics mainScanJobStatistics {};
    CalcSubScanStats(scanStatisticsPath, false, mainScanJobStatistics);
    ReportJobDetailsWithLabel(SubJobStatus::RUNNING,
        OBSPlugin::PROGRESS0,
        "object_storage_plugin_restore_scan_inprogress_label",
        JobLogLevel::TASK_LOG_INFO,
        std::to_string(mainScanJobStatistics.m_totFilesToBackup),
        FormatCapacity(mainScanJobStatistics.m_totalSizeToBackup));
    return true;
}

void ObjectStorageRestoreJob::ReportJobDetailsWithDetailAndErrcode(
    const std::tuple<JobLogLevel::type, SubJobStatus::type, const int> &reportInfo, const std::string& logLabel,
    const int64_t errCode, const std::string& message)
{
    JobLogLevel::type logLevel = std::get<0>(reportInfo);
    SubJobStatus::type jobStatus = std::get<1>(reportInfo);
    int jobProgress = std::get<2>(reportInfo);
    SubJobDetails subJobDetails;
    LogDetail logDetail {};
    logDetail.__set_additionalDesc(vector<std::string>{message});
    std::vector<LogDetail> logDetailList;
    ActionResult result;
    AddLogDetail(logDetail, logLabel, logLevel);
    AddErrCode(logDetail, errCode);
    REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, jobProgress, 0, jobStatus);
}
}